diff options
| author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
|---|---|---|
| committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
| commit | 19fcec84d8d7d21e796c7624e521b60d28ee21ed (patch) | |
| tree | 42d26aa27d1e3f7c0b8bd3fd14e7d7082f5008dc /src/jaegertracing/thrift/lib/netcore/Thrift | |
| parent | Initial commit. (diff) | |
| download | ceph-upstream.tar.xz ceph-upstream.zip | |
Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/jaegertracing/thrift/lib/netcore/Thrift')
55 files changed, 7711 insertions, 0 deletions
diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/TCollections.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/TCollections.cs new file mode 100644 index 000000000..147bfc7d3 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/TCollections.cs @@ -0,0 +1,101 @@ +// 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.Collections; + +namespace Thrift.Collections +{ + // ReSharper disable once InconsistentNaming + public class TCollections + { + /// <summary> + /// This will return true if the two collections are value-wise the same. + /// If the collection contains a collection, the collections will be compared using this method. + /// </summary> + public static bool Equals(IEnumerable first, IEnumerable second) + { + if (first == null && second == null) + { + return true; + } + + if (first == null || second == null) + { + return false; + } + + var fiter = first.GetEnumerator(); + var siter = second.GetEnumerator(); + + var fnext = fiter.MoveNext(); + var snext = siter.MoveNext(); + + while (fnext && snext) + { + var fenum = fiter.Current as IEnumerable; + var senum = siter.Current as IEnumerable; + + if (fenum != null && senum != null) + { + if (!Equals(fenum, senum)) + { + return false; + } + } + else if (fenum == null ^ senum == null) + { + return false; + } + else if (!Equals(fiter.Current, siter.Current)) + { + return false; + } + + fnext = fiter.MoveNext(); + snext = siter.MoveNext(); + } + + return fnext == snext; + } + + /// <summary> + /// This returns a hashcode based on the value of the enumerable. + /// </summary> + public static int GetHashCode(IEnumerable enumerable) + { + if (enumerable == null) + { + return 0; + } + + var hashcode = 0; + + foreach (var obj in enumerable) + { + var enum2 = obj as IEnumerable; + var objHash = enum2 == null ? obj.GetHashCode() : GetHashCode(enum2); + + unchecked + { + hashcode = (hashcode*397) ^ (objHash); + } + } + + return hashcode; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/THashSet.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/THashSet.cs new file mode 100644 index 000000000..011f0a0d6 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Collections/THashSet.cs @@ -0,0 +1,67 @@ +// 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.Collections; +using System.Collections.Generic; + +namespace Thrift.Collections +{ + // ReSharper disable once InconsistentNaming + public class THashSet<T> : ICollection<T> + { + private readonly HashSet<T> _set = new HashSet<T>(); + + public int Count => _set.Count; + + public bool IsReadOnly => false; + + public void Add(T item) + { + _set.Add(item); + } + + public void Clear() + { + _set.Clear(); + } + + public bool Contains(T item) + { + return _set.Contains(item); + } + + public void CopyTo(T[] array, int arrayIndex) + { + _set.CopyTo(array, arrayIndex); + } + + public IEnumerator GetEnumerator() + { + return _set.GetEnumerator(); + } + + IEnumerator<T> IEnumerable<T>.GetEnumerator() + { + return ((IEnumerable<T>) _set).GetEnumerator(); + } + + public bool Remove(T item) + { + return _set.Remove(item); + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/ITAsyncProcessor.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/ITAsyncProcessor.cs new file mode 100644 index 000000000..db8e40aef --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/ITAsyncProcessor.cs @@ -0,0 +1,29 @@ +// 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; +using Thrift.Protocols; + +namespace Thrift +{ + public interface ITAsyncProcessor + { + Task<bool> ProcessAsync(TProtocol iprot, TProtocol oprot); + Task<bool> ProcessAsync(TProtocol iprot, TProtocol oprot, CancellationToken cancellationToken); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/ITProcessorFactory.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/ITProcessorFactory.cs new file mode 100644 index 000000000..5133e5c48 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/ITProcessorFactory.cs @@ -0,0 +1,28 @@ +// 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 Thrift.Server; +using Thrift.Transports; + +namespace Thrift +{ + // ReSharper disable once InconsistentNaming + public interface ITProcessorFactory + { + ITAsyncProcessor GetAsyncProcessor(TClientTransport trans, TBaseServer baseServer = null); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Properties/AssemblyInfo.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Properties/AssemblyInfo.cs new file mode 100644 index 000000000..0f433b600 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Properties/AssemblyInfo.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 System.Reflection; +using System.Runtime.InteropServices; + +// General Information about an assembly is controlled through the following +// set of attributes. Change these attribute values to modify the information +// associated with an assembly. + +[assembly: AssemblyTitle("Thrift")] +[assembly: AssemblyDescription("C# .NET Core bindings for the Apache Thrift RPC system")] +[assembly: AssemblyConfiguration("")] +[assembly: AssemblyCompany("The Apache Software Foundation")] +[assembly: AssemblyProduct("Thrift")] +[assembly: AssemblyCopyright("The Apache Software Foundation")] +[assembly: AssemblyTrademark("")] +[assembly: AssemblyCulture("")] +//@TODO where to put License information? + +// Setting ComVisible to false makes the types in this assembly not visible +// to COM components. If you need to access a exType in this assembly from +// COM, set the ComVisible attribute to true on that exType. + +[assembly: ComVisible(false)] + +// The following GUID is for the ID of the typelib if this project is exposed to COM + +[assembly: Guid("df3f8ef0-e0a3-4c86-a65b-8ec84e016b1d")] + +// Version information for an assembly consists of the following four values: +// +// Major Version +// Minor Version +// Build Number +// Revision +// +// You can specify all the values or you can default the Build and Revision Numbers +// by using the '*' as shown below: + +[assembly: AssemblyVersion("0.13.0.0")] +[assembly: AssemblyFileVersion("0.13.0.0")] diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TField.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TField.cs new file mode 100644 index 000000000..d311535e7 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TField.cs @@ -0,0 +1,37 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TField + { + public TField(string name, TType type, short id) + { + Name = name; + Type = type; + ID = id; + } + + public string Name { get; set; } + + public TType Type { get; set; } + + // ReSharper disable once InconsistentNaming - do not rename - it used for generation + public short ID { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TList.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TList.cs new file mode 100644 index 000000000..ce232207c --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TList.cs @@ -0,0 +1,33 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TList + { + public TList(TType elementType, int count) + { + ElementType = elementType; + Count = count; + } + + public TType ElementType { get; set; } + + public int Count { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMap.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMap.cs new file mode 100644 index 000000000..9195593db --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMap.cs @@ -0,0 +1,36 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TMap + { + public TMap(TType keyType, TType valueType, int count) + { + KeyType = keyType; + ValueType = valueType; + Count = count; + } + + public TType KeyType { get; set; } + + public TType ValueType { get; set; } + + public int Count { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessage.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessage.cs new file mode 100644 index 000000000..17f49298f --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessage.cs @@ -0,0 +1,37 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TMessage + { + public TMessage(string name, TMessageType type, int seqid) + { + Name = name; + Type = type; + SeqID = seqid; + } + + public string Name { get; set; } + + public TMessageType Type { get; set; } + + // ReSharper disable once InconsistentNaming - do not rename - it used for generation + public int SeqID { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessageType.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessageType.cs new file mode 100644 index 000000000..d7b9a2275 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TMessageType.cs @@ -0,0 +1,28 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public enum TMessageType + { + Call = 1, + Reply = 2, + Exception = 3, + Oneway = 4 + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TSet.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TSet.cs new file mode 100644 index 000000000..a583b54a6 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TSet.cs @@ -0,0 +1,38 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TSet + { + public TSet(TType elementType, int count) + { + ElementType = elementType; + Count = count; + } + + public TSet(TList list) + : this(list.ElementType, list.Count) + { + } + + public TType ElementType { get; set; } + + public int Count { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TStruct.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TStruct.cs new file mode 100644 index 000000000..a28dcc3d0 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TStruct.cs @@ -0,0 +1,30 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public struct TStruct + { + public TStruct(string name) + { + Name = name; + } + + public string Name { get; set; } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TType.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TType.cs new file mode 100644 index 000000000..ebe781c95 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Entities/TType.cs @@ -0,0 +1,37 @@ +// 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.Protocols.Entities +{ + // ReSharper disable once InconsistentNaming + public enum TType : byte + { + Stop = 0, + Void = 1, + Bool = 2, + Byte = 3, + Double = 4, + I16 = 6, + I32 = 8, + I64 = 10, + String = 11, + Struct = 12, + Map = 13, + Set = 14, + List = 15 + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/ITProtocolFactory.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/ITProtocolFactory.cs new file mode 100644 index 000000000..ecc5cc494 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/ITProtocolFactory.cs @@ -0,0 +1,27 @@ +// 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 Thrift.Transports; + +namespace Thrift.Protocols +{ + // ReSharper disable once InconsistentNaming + public interface ITProtocolFactory + { + TProtocol GetProtocol(TClientTransport trans); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TAbstractBase.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TAbstractBase.cs new file mode 100644 index 000000000..4e18681bf --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TAbstractBase.cs @@ -0,0 +1,28 @@ +// 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.Protocols +{ + // ReSharper disable once InconsistentNaming + public interface TAbstractBase + { + Task WriteAsync(TProtocol tProtocol, CancellationToken cancellationToken); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBase.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBase.cs new file mode 100644 index 000000000..014e1aee8 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBase.cs @@ -0,0 +1,28 @@ +// 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.Protocols +{ + // ReSharper disable once InconsistentNaming + public interface TBase : TAbstractBase + { + Task ReadAsync(TProtocol tProtocol, CancellationToken cancellationToken); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBinaryProtocol.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBinaryProtocol.cs new file mode 100644 index 000000000..deec85c42 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TBinaryProtocol.cs @@ -0,0 +1,613 @@ +// 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 Thrift.Protocols.Entities; +using Thrift.Transports; + +namespace Thrift.Protocols +{ + // ReSharper disable once InconsistentNaming + public class TBinaryProtocol : TProtocol + { + //TODO: Unit tests + //TODO: Localization + //TODO: pragma + + protected const uint VersionMask = 0xffff0000; + protected const uint Version1 = 0x80010000; + + protected bool StrictRead; + protected bool StrictWrite; + + public TBinaryProtocol(TClientTransport trans) + : this(trans, false, true) + { + } + + public TBinaryProtocol(TClientTransport trans, bool strictRead, bool strictWrite) + : base(trans) + { + StrictRead = strictRead; + StrictWrite = strictWrite; + } + + public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + if (StrictWrite) + { + var version = Version1 | (uint) message.Type; + await WriteI32Async((int) version, cancellationToken); + await WriteStringAsync(message.Name, cancellationToken); + await WriteI32Async(message.SeqID, cancellationToken); + } + else + { + await WriteStringAsync(message.Name, cancellationToken); + await WriteByteAsync((sbyte) message.Type, cancellationToken); + await WriteI32Async(message.SeqID, cancellationToken); + } + } + + public override async Task WriteMessageEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteStructBeginAsync(TStruct @struct, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteStructEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteFieldBeginAsync(TField field, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync((sbyte) field.Type, cancellationToken); + await WriteI16Async(field.ID, cancellationToken); + } + + public override async Task WriteFieldEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteFieldStopAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync((sbyte) TType.Stop, cancellationToken); + } + + public override async Task WriteMapBeginAsync(TMap map, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync((sbyte) map.KeyType, cancellationToken); + await WriteByteAsync((sbyte) map.ValueType, cancellationToken); + await WriteI32Async(map.Count, cancellationToken); + } + + public override async Task WriteMapEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteListBeginAsync(TList list, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync((sbyte) list.ElementType, cancellationToken); + await WriteI32Async(list.Count, cancellationToken); + } + + public override async Task WriteListEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteSetBeginAsync(TSet set, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync((sbyte) set.ElementType, cancellationToken); + await WriteI32Async(set.Count, cancellationToken); + } + + public override async Task WriteSetEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteBoolAsync(bool b, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteByteAsync(b ? (sbyte) 1 : (sbyte) 0, cancellationToken); + } + + protected internal static byte[] CreateWriteByte(sbyte b) + { + var bout = new byte[1]; + + bout[0] = (byte) b; + + return bout; + } + + public override async Task WriteByteAsync(sbyte b, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bout = CreateWriteByte(b); + await Trans.WriteAsync(bout, 0, 1, cancellationToken); + } + + protected internal static byte[] CreateWriteI16(short s) + { + var i16Out = new byte[2]; + + i16Out[0] = (byte) (0xff & (s >> 8)); + i16Out[1] = (byte) (0xff & s); + + return i16Out; + } + + public override async Task WriteI16Async(short i16, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var i16Out = CreateWriteI16(i16); + await Trans.WriteAsync(i16Out, 0, 2, cancellationToken); + } + + protected internal static byte[] CreateWriteI32(int i32) + { + var i32Out = new byte[4]; + + i32Out[0] = (byte) (0xff & (i32 >> 24)); + i32Out[1] = (byte) (0xff & (i32 >> 16)); + i32Out[2] = (byte) (0xff & (i32 >> 8)); + i32Out[3] = (byte) (0xff & i32); + + return i32Out; + } + + public override async Task WriteI32Async(int i32, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var i32Out = CreateWriteI32(i32); + await Trans.WriteAsync(i32Out, 0, 4, cancellationToken); + } + + protected internal static byte[] CreateWriteI64(long i64) + { + var i64Out = new byte[8]; + + i64Out[0] = (byte) (0xff & (i64 >> 56)); + i64Out[1] = (byte) (0xff & (i64 >> 48)); + i64Out[2] = (byte) (0xff & (i64 >> 40)); + i64Out[3] = (byte) (0xff & (i64 >> 32)); + i64Out[4] = (byte) (0xff & (i64 >> 24)); + i64Out[5] = (byte) (0xff & (i64 >> 16)); + i64Out[6] = (byte) (0xff & (i64 >> 8)); + i64Out[7] = (byte) (0xff & i64); + + return i64Out; + } + + public override async Task WriteI64Async(long i64, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var i64Out = CreateWriteI64(i64); + await Trans.WriteAsync(i64Out, 0, 8, cancellationToken); + } + + public override async Task WriteDoubleAsync(double d, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteI64Async(BitConverter.DoubleToInt64Bits(d), cancellationToken); + } + + public override async Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteI32Async(bytes.Length, cancellationToken); + await Trans.WriteAsync(bytes, 0, bytes.Length, cancellationToken); + } + + public override async Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TMessage>(cancellationToken); + } + + var message = new TMessage(); + var size = await ReadI32Async(cancellationToken); + if (size < 0) + { + var version = (uint) size & VersionMask; + if (version != Version1) + { + throw new TProtocolException(TProtocolException.BAD_VERSION, + $"Bad version in ReadMessageBegin: {version}"); + } + message.Type = (TMessageType) (size & 0x000000ff); + message.Name = await ReadStringAsync(cancellationToken); + message.SeqID = await ReadI32Async(cancellationToken); + } + else + { + if (StrictRead) + { + throw new TProtocolException(TProtocolException.BAD_VERSION, + "Missing version in ReadMessageBegin, old client?"); + } + message.Name = await ReadStringBodyAsync(size, cancellationToken); + message.Type = (TMessageType) await ReadByteAsync(cancellationToken); + message.SeqID = await ReadI32Async(cancellationToken); + } + return message; + } + + public override async Task ReadMessageEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TStruct> ReadStructBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + //TODO: no read from internal transport? + return new TStruct(); + } + + public override async Task ReadStructEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TField> ReadFieldBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TField>(cancellationToken); + } + + var field = new TField + { + Type = (TType) await ReadByteAsync(cancellationToken) + }; + + if (field.Type != TType.Stop) + { + field.ID = await ReadI16Async(cancellationToken); + } + + return field; + } + + public override async Task ReadFieldEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TMap> ReadMapBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TMap>(cancellationToken); + } + + var map = new TMap + { + KeyType = (TType) await ReadByteAsync(cancellationToken), + ValueType = (TType) await ReadByteAsync(cancellationToken), + Count = await ReadI32Async(cancellationToken) + }; + + return map; + } + + public override async Task ReadMapEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TList> ReadListBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TList>(cancellationToken); + } + + var list = new TList + { + ElementType = (TType) await ReadByteAsync(cancellationToken), + Count = await ReadI32Async(cancellationToken) + }; + + return list; + } + + public override async Task ReadListEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TSet> ReadSetBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TSet>(cancellationToken); + } + + var set = new TSet + { + ElementType = (TType) await ReadByteAsync(cancellationToken), + Count = await ReadI32Async(cancellationToken) + }; + + return set; + } + + public override async Task ReadSetEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<bool> ReadBoolAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<bool>(cancellationToken); + } + + return await ReadByteAsync(cancellationToken) == 1; + } + + public override async Task<sbyte> ReadByteAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<sbyte>(cancellationToken); + } + + var bin = new byte[1]; + await Trans.ReadAllAsync(bin, 0, 1, cancellationToken); //TODO: why readall ? + return (sbyte) bin[0]; + } + + public override async Task<short> ReadI16Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<short>(cancellationToken); + } + + var i16In = new byte[2]; + await Trans.ReadAllAsync(i16In, 0, 2, cancellationToken); + var result = (short) (((i16In[0] & 0xff) << 8) | i16In[1] & 0xff); + return result; + } + + public override async Task<int> ReadI32Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<int>(cancellationToken); + } + + var i32In = new byte[4]; + await Trans.ReadAllAsync(i32In, 0, 4, cancellationToken); + + var result = + ((i32In[0] & 0xff) << 24) | + ((i32In[1] & 0xff) << 16) | + ((i32In[2] & 0xff) << 8) | + i32In[3] & 0xff; + + return result; + } + +#pragma warning disable 675 + + protected internal long CreateReadI64(byte[] buf) + { + var result = + ((long) (buf[0] & 0xff) << 56) | + ((long) (buf[1] & 0xff) << 48) | + ((long) (buf[2] & 0xff) << 40) | + ((long) (buf[3] & 0xff) << 32) | + ((long) (buf[4] & 0xff) << 24) | + ((long) (buf[5] & 0xff) << 16) | + ((long) (buf[6] & 0xff) << 8) | + buf[7] & 0xff; + + return result; + } + +#pragma warning restore 675 + + public override async Task<long> ReadI64Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<long>(cancellationToken); + } + + var i64In = new byte[8]; + await Trans.ReadAllAsync(i64In, 0, 8, cancellationToken); + return CreateReadI64(i64In); + } + + public override async Task<double> ReadDoubleAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<double>(cancellationToken); + } + + var d = await ReadI64Async(cancellationToken); + return BitConverter.Int64BitsToDouble(d); + } + + public override async Task<byte[]> ReadBinaryAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<byte[]>(cancellationToken); + } + + var size = await ReadI32Async(cancellationToken); + var buf = new byte[size]; + await Trans.ReadAllAsync(buf, 0, size, cancellationToken); + return buf; + } + + private async Task<string> ReadStringBodyAsync(int size, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled<string>(cancellationToken); + } + + var buf = new byte[size]; + await Trans.ReadAllAsync(buf, 0, size, cancellationToken); + return Encoding.UTF8.GetString(buf, 0, buf.Length); + } + + public class Factory : ITProtocolFactory + { + protected bool StrictRead; + protected bool StrictWrite; + + public Factory() + : this(false, true) + { + } + + public Factory(bool strictRead, bool strictWrite) + { + StrictRead = strictRead; + StrictWrite = strictWrite; + } + + public TProtocol GetProtocol(TClientTransport trans) + { + return new TBinaryProtocol(trans, StrictRead, StrictWrite); + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TCompactProtocol.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TCompactProtocol.cs new file mode 100644 index 000000000..cecdf03cc --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TCompactProtocol.cs @@ -0,0 +1,922 @@ +// 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.Text; +using System.Threading; +using System.Threading.Tasks; +using Thrift.Protocols.Entities; +using Thrift.Transports; + +namespace Thrift.Protocols +{ + //TODO: implementation of TProtocol + + // ReSharper disable once InconsistentNaming + public class TCompactProtocol : TProtocol + { + private const byte ProtocolId = 0x82; + private const byte Version = 1; + private const byte VersionMask = 0x1f; // 0001 1111 + private const byte TypeMask = 0xE0; // 1110 0000 + private const byte TypeBits = 0x07; // 0000 0111 + private const int TypeShiftAmount = 5; + private static readonly TStruct AnonymousStruct = new TStruct(string.Empty); + private static readonly TField Tstop = new TField(string.Empty, TType.Stop, 0); + + // ReSharper disable once InconsistentNaming + private static readonly byte[] TTypeToCompactType = new byte[16]; + + /// <summary> + /// Used to keep track of the last field for the current and previous structs, so we can do the delta stuff. + /// </summary> + private readonly Stack<short> _lastField = new Stack<short>(15); + + /// <summary> + /// If we encounter a boolean field begin, save the TField here so it can have the value incorporated. + /// </summary> + private TField? _booleanField; + + /// <summary> + /// If we Read a field header, and it's a boolean field, save the boolean value here so that ReadBool can use it. + /// </summary> + private bool? _boolValue; + + private short _lastFieldId; + + public TCompactProtocol(TClientTransport trans) + : base(trans) + { + TTypeToCompactType[(int) TType.Stop] = Types.Stop; + TTypeToCompactType[(int) TType.Bool] = Types.BooleanTrue; + TTypeToCompactType[(int) TType.Byte] = Types.Byte; + TTypeToCompactType[(int) TType.I16] = Types.I16; + TTypeToCompactType[(int) TType.I32] = Types.I32; + TTypeToCompactType[(int) TType.I64] = Types.I64; + TTypeToCompactType[(int) TType.Double] = Types.Double; + TTypeToCompactType[(int) TType.String] = Types.Binary; + TTypeToCompactType[(int) TType.List] = Types.List; + TTypeToCompactType[(int) TType.Set] = Types.Set; + TTypeToCompactType[(int) TType.Map] = Types.Map; + TTypeToCompactType[(int) TType.Struct] = Types.Struct; + } + + public void Reset() + { + _lastField.Clear(); + _lastFieldId = 0; + } + + public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await Trans.WriteAsync(new[] {ProtocolId}, cancellationToken); + await + Trans.WriteAsync( + new[] {(byte) ((Version & VersionMask) | (((uint) message.Type << TypeShiftAmount) & TypeMask))}, + cancellationToken); + + var bufferTuple = CreateWriteVarInt32((uint) message.SeqID); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + + await WriteStringAsync(message.Name, cancellationToken); + } + + public override async Task WriteMessageEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + /// <summary> + /// Write a struct begin. This doesn't actually put anything on the wire. We + /// use it as an opportunity to put special placeholder markers on the field + /// stack so we can get the field id deltas correct. + /// </summary> + public override async Task WriteStructBeginAsync(TStruct @struct, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + _lastField.Push(_lastFieldId); + _lastFieldId = 0; + } + + public override async Task WriteStructEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + _lastFieldId = _lastField.Pop(); + } + + private async Task WriteFieldBeginInternalAsync(TField field, byte typeOverride, + CancellationToken cancellationToken) + { + // if there's a exType override, use that. + var typeToWrite = typeOverride == 0xFF ? GetCompactType(field.Type) : typeOverride; + + // check if we can use delta encoding for the field id + if ((field.ID > _lastFieldId) && (field.ID - _lastFieldId <= 15)) + { + var b = (byte) (((field.ID - _lastFieldId) << 4) | typeToWrite); + // Write them together + await Trans.WriteAsync(new[] {b}, cancellationToken); + } + else + { + // Write them separate + await Trans.WriteAsync(new[] {typeToWrite}, cancellationToken); + await WriteI16Async(field.ID, cancellationToken); + } + + _lastFieldId = field.ID; + } + + public override async Task WriteFieldBeginAsync(TField field, CancellationToken cancellationToken) + { + if (field.Type == TType.Bool) + { + _booleanField = field; + } + else + { + await WriteFieldBeginInternalAsync(field, 0xFF, cancellationToken); + } + } + + public override async Task WriteFieldEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteFieldStopAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await Trans.WriteAsync(new[] {Types.Stop}, cancellationToken); + } + + protected async Task WriteCollectionBeginAsync(TType elemType, int size, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + /* + Abstract method for writing the start of lists and sets. List and sets on + the wire differ only by the exType indicator. + */ + + if (size <= 14) + { + await Trans.WriteAsync(new[] {(byte) ((size << 4) | GetCompactType(elemType))}, cancellationToken); + } + else + { + await Trans.WriteAsync(new[] {(byte) (0xf0 | GetCompactType(elemType))}, cancellationToken); + + var bufferTuple = CreateWriteVarInt32((uint) size); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + } + } + + public override async Task WriteListBeginAsync(TList list, CancellationToken cancellationToken) + { + await WriteCollectionBeginAsync(list.ElementType, list.Count, cancellationToken); + } + + public override async Task WriteListEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteSetBeginAsync(TSet set, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await WriteCollectionBeginAsync(set.ElementType, set.Count, cancellationToken); + } + + public override async Task WriteSetEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteBoolAsync(bool b, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + /* + Write a boolean value. Potentially, this could be a boolean field, in + which case the field header info isn't written yet. If so, decide what the + right exType header is for the value and then Write the field header. + Otherwise, Write a single byte. + */ + + if (_booleanField != null) + { + // we haven't written the field header yet + await + WriteFieldBeginInternalAsync(_booleanField.Value, b ? Types.BooleanTrue : Types.BooleanFalse, + cancellationToken); + _booleanField = null; + } + else + { + // we're not part of a field, so just Write the value. + await Trans.WriteAsync(new[] {b ? Types.BooleanTrue : Types.BooleanFalse}, cancellationToken); + } + } + + public override async Task WriteByteAsync(sbyte b, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await Trans.WriteAsync(new[] {(byte) b}, cancellationToken); + } + + public override async Task WriteI16Async(short i16, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bufferTuple = CreateWriteVarInt32(IntToZigzag(i16)); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + } + + protected internal Tuple<byte[], int> CreateWriteVarInt32(uint n) + { + // Write an i32 as a varint.Results in 1 - 5 bytes on the wire. + var i32Buf = new byte[5]; + var idx = 0; + + while (true) + { + if ((n & ~0x7F) == 0) + { + i32Buf[idx++] = (byte) n; + break; + } + + i32Buf[idx++] = (byte) ((n & 0x7F) | 0x80); + n >>= 7; + } + + return new Tuple<byte[], int>(i32Buf, idx); + } + + public override async Task WriteI32Async(int i32, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bufferTuple = CreateWriteVarInt32(IntToZigzag(i32)); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + } + + protected internal Tuple<byte[], int> CreateWriteVarInt64(ulong n) + { + // Write an i64 as a varint. Results in 1-10 bytes on the wire. + var buf = new byte[10]; + var idx = 0; + + while (true) + { + if ((n & ~(ulong) 0x7FL) == 0) + { + buf[idx++] = (byte) n; + break; + } + buf[idx++] = (byte) ((n & 0x7F) | 0x80); + n >>= 7; + } + + return new Tuple<byte[], int>(buf, idx); + } + + public override async Task WriteI64Async(long i64, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bufferTuple = CreateWriteVarInt64(LongToZigzag(i64)); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + } + + public override async Task WriteDoubleAsync(double d, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var data = new byte[8]; + FixedLongToBytes(BitConverter.DoubleToInt64Bits(d), data, 0); + await Trans.WriteAsync(data, cancellationToken); + } + + public override async Task WriteStringAsync(string str, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bytes = Encoding.UTF8.GetBytes(str); + + var bufferTuple = CreateWriteVarInt32((uint) bytes.Length); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + await Trans.WriteAsync(bytes, 0, bytes.Length, cancellationToken); + } + + public override async Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + var bufferTuple = CreateWriteVarInt32((uint) bytes.Length); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + await Trans.WriteAsync(bytes, 0, bytes.Length, cancellationToken); + } + + public override async Task WriteMapBeginAsync(TMap map, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + if (map.Count == 0) + { + await Trans.WriteAsync(new[] {(byte) 0}, cancellationToken); + } + else + { + var bufferTuple = CreateWriteVarInt32((uint) map.Count); + await Trans.WriteAsync(bufferTuple.Item1, 0, bufferTuple.Item2, cancellationToken); + await + Trans.WriteAsync( + new[] {(byte) ((GetCompactType(map.KeyType) << 4) | GetCompactType(map.ValueType))}, + cancellationToken); + } + } + + public override async Task WriteMapEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TMessage>(cancellationToken); + } + + var protocolId = (byte) await ReadByteAsync(cancellationToken); + if (protocolId != ProtocolId) + { + throw new TProtocolException($"Expected protocol id {ProtocolId:X} but got {protocolId:X}"); + } + + var versionAndType = (byte) await ReadByteAsync(cancellationToken); + var version = (byte) (versionAndType & VersionMask); + + if (version != Version) + { + throw new TProtocolException($"Expected version {Version} but got {version}"); + } + + var type = (byte) ((versionAndType >> TypeShiftAmount) & TypeBits); + var seqid = (int) await ReadVarInt32Async(cancellationToken); + var messageName = await ReadStringAsync(cancellationToken); + + return new TMessage(messageName, (TMessageType) type, seqid); + } + + public override async Task ReadMessageEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TStruct> ReadStructBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TStruct>(cancellationToken); + } + + // some magic is here ) + + _lastField.Push(_lastFieldId); + _lastFieldId = 0; + + return AnonymousStruct; + } + + public override async Task ReadStructEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + /* + Doesn't actually consume any wire data, just removes the last field for + this struct from the field stack. + */ + + // consume the last field we Read off the wire. + _lastFieldId = _lastField.Pop(); + } + + public override async Task<TField> ReadFieldBeginAsync(CancellationToken cancellationToken) + { + // Read a field header off the wire. + var type = (byte) await ReadByteAsync(cancellationToken); + // if it's a stop, then we can return immediately, as the struct is over. + if (type == Types.Stop) + { + return Tstop; + } + + short fieldId; + // mask off the 4 MSB of the exType header. it could contain a field id delta. + var modifier = (short) ((type & 0xf0) >> 4); + if (modifier == 0) + { + fieldId = await ReadI16Async(cancellationToken); + } + else + { + fieldId = (short) (_lastFieldId + modifier); + } + + var field = new TField(string.Empty, GetTType((byte) (type & 0x0f)), fieldId); + // if this happens to be a boolean field, the value is encoded in the exType + if (IsBoolType(type)) + { + _boolValue = (byte) (type & 0x0f) == Types.BooleanTrue; + } + + // push the new field onto the field stack so we can keep the deltas going. + _lastFieldId = field.ID; + return field; + } + + public override async Task ReadFieldEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TMap> ReadMapBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled<TMap>(cancellationToken); + } + + /* + Read a map header off the wire. If the size is zero, skip Reading the key + and value exType. This means that 0-length maps will yield TMaps without the + "correct" types. + */ + + var size = (int) await ReadVarInt32Async(cancellationToken); + var keyAndValueType = size == 0 ? (byte) 0 : (byte) await ReadByteAsync(cancellationToken); + return new TMap(GetTType((byte) (keyAndValueType >> 4)), GetTType((byte) (keyAndValueType & 0xf)), size); + } + + public override async Task ReadMapEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task<TSet> ReadSetBeginAsync(CancellationToken cancellationToken) + { + /* + Read a set header off the wire. If the set size is 0-14, the size will + be packed into the element exType header. If it's a longer set, the 4 MSB + of the element exType header will be 0xF, and a varint will follow with the + true size. + */ + + return new TSet(await ReadListBeginAsync(cancellationToken)); + } + + public override async Task<bool> ReadBoolAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<bool>(cancellationToken); + } + + /* + Read a boolean off the wire. If this is a boolean field, the value should + already have been Read during ReadFieldBegin, so we'll just consume the + pre-stored value. Otherwise, Read a byte. + */ + + if (_boolValue != null) + { + var result = _boolValue.Value; + _boolValue = null; + return result; + } + + return await ReadByteAsync(cancellationToken) == Types.BooleanTrue; + } + + public override async Task<sbyte> ReadByteAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<sbyte>(cancellationToken); + } + + // Read a single byte off the wire. Nothing interesting here. + var buf = new byte[1]; + await Trans.ReadAllAsync(buf, 0, 1, cancellationToken); + return (sbyte) buf[0]; + } + + public override async Task<short> ReadI16Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<short>(cancellationToken); + } + + return (short) ZigzagToInt(await ReadVarInt32Async(cancellationToken)); + } + + public override async Task<int> ReadI32Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<int>(cancellationToken); + } + + return ZigzagToInt(await ReadVarInt32Async(cancellationToken)); + } + + public override async Task<long> ReadI64Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<long>(cancellationToken); + } + + return ZigzagToLong(await ReadVarInt64Async(cancellationToken)); + } + + public override async Task<double> ReadDoubleAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<double>(cancellationToken); + } + + var longBits = new byte[8]; + await Trans.ReadAllAsync(longBits, 0, 8, cancellationToken); + + return BitConverter.Int64BitsToDouble(BytesToLong(longBits)); + } + + public override async Task<string> ReadStringAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled<string>(cancellationToken); + } + + // Reads a byte[] (via ReadBinary), and then UTF-8 decodes it. + var length = (int) await ReadVarInt32Async(cancellationToken); + + if (length == 0) + { + return string.Empty; + } + + var buf = new byte[length]; + await Trans.ReadAllAsync(buf, 0, length, cancellationToken); + + return Encoding.UTF8.GetString(buf); + } + + public override async Task<byte[]> ReadBinaryAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<byte[]>(cancellationToken); + } + + // Read a byte[] from the wire. + var length = (int) await ReadVarInt32Async(cancellationToken); + if (length == 0) + { + return new byte[0]; + } + + var buf = new byte[length]; + await Trans.ReadAllAsync(buf, 0, length, cancellationToken); + return buf; + } + + public override async Task<TList> ReadListBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled<TList>(cancellationToken); + } + + /* + Read a list header off the wire. If the list size is 0-14, the size will + be packed into the element exType header. If it's a longer list, the 4 MSB + of the element exType header will be 0xF, and a varint will follow with the + true size. + */ + + var sizeAndType = (byte) await ReadByteAsync(cancellationToken); + var size = (sizeAndType >> 4) & 0x0f; + if (size == 15) + { + size = (int) await ReadVarInt32Async(cancellationToken); + } + + var type = GetTType(sizeAndType); + return new TList(type, size); + } + + public override async Task ReadListEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task ReadSetEndAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + private static byte GetCompactType(TType ttype) + { + // Given a TType value, find the appropriate TCompactProtocol.Types constant. + return TTypeToCompactType[(int) ttype]; + } + + + private async Task<uint> ReadVarInt32Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<uint>(cancellationToken); + } + + /* + Read an i32 from the wire as a varint. The MSB of each byte is set + if there is another byte to follow. This can Read up to 5 bytes. + */ + + uint result = 0; + var shift = 0; + + while (true) + { + var b = (byte) await ReadByteAsync(cancellationToken); + result |= (uint) (b & 0x7f) << shift; + if ((b & 0x80) != 0x80) + { + break; + } + shift += 7; + } + + return result; + } + + private async Task<ulong> ReadVarInt64Async(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<uint>(cancellationToken); + } + + /* + Read an i64 from the wire as a proper varint. The MSB of each byte is set + if there is another byte to follow. This can Read up to 10 bytes. + */ + + var shift = 0; + ulong result = 0; + while (true) + { + var b = (byte) await ReadByteAsync(cancellationToken); + result |= (ulong) (b & 0x7f) << shift; + if ((b & 0x80) != 0x80) + { + break; + } + shift += 7; + } + + return result; + } + + private static int ZigzagToInt(uint n) + { + return (int) (n >> 1) ^ -(int) (n & 1); + } + + private static long ZigzagToLong(ulong n) + { + return (long) (n >> 1) ^ -(long) (n & 1); + } + + private static long BytesToLong(byte[] bytes) + { + /* + Note that it's important that the mask bytes are long literals, + otherwise they'll default to ints, and when you shift an int left 56 bits, + you just get a messed up int. + */ + + return + ((bytes[7] & 0xffL) << 56) | + ((bytes[6] & 0xffL) << 48) | + ((bytes[5] & 0xffL) << 40) | + ((bytes[4] & 0xffL) << 32) | + ((bytes[3] & 0xffL) << 24) | + ((bytes[2] & 0xffL) << 16) | + ((bytes[1] & 0xffL) << 8) | + (bytes[0] & 0xffL); + } + + private static bool IsBoolType(byte b) + { + var lowerNibble = b & 0x0f; + return (lowerNibble == Types.BooleanTrue) || (lowerNibble == Types.BooleanFalse); + } + + private static TType GetTType(byte type) + { + // Given a TCompactProtocol.Types constant, convert it to its corresponding TType value. + switch ((byte) (type & 0x0f)) + { + case Types.Stop: + return TType.Stop; + case Types.BooleanFalse: + case Types.BooleanTrue: + return TType.Bool; + case Types.Byte: + return TType.Byte; + case Types.I16: + return TType.I16; + case Types.I32: + return TType.I32; + case Types.I64: + return TType.I64; + case Types.Double: + return TType.Double; + case Types.Binary: + return TType.String; + case Types.List: + return TType.List; + case Types.Set: + return TType.Set; + case Types.Map: + return TType.Map; + case Types.Struct: + return TType.Struct; + default: + throw new TProtocolException($"Don't know what exType: {(byte) (type & 0x0f)}"); + } + } + + private static ulong LongToZigzag(long n) + { + // Convert l into a zigzag long. This allows negative numbers to be represented compactly as a varint + return (ulong) (n << 1) ^ (ulong) (n >> 63); + } + + private static uint IntToZigzag(int n) + { + // Convert n into a zigzag int. This allows negative numbers to be represented compactly as a varint + return (uint) (n << 1) ^ (uint) (n >> 31); + } + + private static void FixedLongToBytes(long n, byte[] buf, int off) + { + // Convert a long into little-endian bytes in buf starting at off and going until off+7. + buf[off + 0] = (byte) (n & 0xff); + buf[off + 1] = (byte) ((n >> 8) & 0xff); + buf[off + 2] = (byte) ((n >> 16) & 0xff); + buf[off + 3] = (byte) ((n >> 24) & 0xff); + buf[off + 4] = (byte) ((n >> 32) & 0xff); + buf[off + 5] = (byte) ((n >> 40) & 0xff); + buf[off + 6] = (byte) ((n >> 48) & 0xff); + buf[off + 7] = (byte) ((n >> 56) & 0xff); + } + + public class Factory : ITProtocolFactory + { + public TProtocol GetProtocol(TClientTransport trans) + { + return new TCompactProtocol(trans); + } + } + + /// <summary> + /// All of the on-wire exType codes. + /// </summary> + private static class Types + { + public const byte Stop = 0x00; + public const byte BooleanTrue = 0x01; + public const byte BooleanFalse = 0x02; + public const byte Byte = 0x03; + public const byte I16 = 0x04; + public const byte I32 = 0x05; + public const byte I64 = 0x06; + public const byte Double = 0x07; + public const byte Binary = 0x08; + public const byte List = 0x09; + public const byte Set = 0x0A; + public const byte Map = 0x0B; + public const byte Struct = 0x0C; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TJSONProtocol.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TJSONProtocol.cs new file mode 100644 index 000000000..6d33f029e --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TJSONProtocol.cs @@ -0,0 +1,981 @@ +// 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.Globalization; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Thrift.Protocols.Entities; +using Thrift.Protocols.Utilities; +using Thrift.Transports; + +namespace Thrift.Protocols +{ + /// <summary> + /// JSON protocol implementation for thrift. + /// This is a full-featured protocol supporting Write and Read. + /// Please see the C++ class header for a detailed description of the + /// protocol's wire format. + /// Adapted from the Java version. + /// </summary> + // ReSharper disable once InconsistentNaming + public class TJsonProtocol : TProtocol + { + private const long Version = 1; + + // Temporary buffer used by several methods + private readonly byte[] _tempBuffer = new byte[4]; + + // Current context that we are in + protected JSONBaseContext Context; + + // Stack of nested contexts that we may be in + protected Stack<JSONBaseContext> ContextStack = new Stack<JSONBaseContext>(); + + // Reader that manages a 1-byte buffer + protected LookaheadReader Reader; + + // Default encoding + protected Encoding Utf8Encoding = Encoding.UTF8; + + /// <summary> + /// TJsonProtocol Constructor + /// </summary> + public TJsonProtocol(TClientTransport trans) + : base(trans) + { + Context = new JSONBaseContext(this); + Reader = new LookaheadReader(this); + } + + /// <summary> + /// Push a new JSON context onto the stack. + /// </summary> + protected void PushContext(JSONBaseContext c) + { + ContextStack.Push(Context); + Context = c; + } + + /// <summary> + /// Pop the last JSON context off the stack + /// </summary> + protected void PopContext() + { + Context = ContextStack.Pop(); + } + + /// <summary> + /// Read a byte that must match b[0]; otherwise an exception is thrown. + /// Marked protected to avoid synthetic accessor in JSONListContext.Read + /// and JSONPairContext.Read + /// </summary> + protected async Task ReadJsonSyntaxCharAsync(byte[] bytes, CancellationToken cancellationToken) + { + var ch = await Reader.ReadAsync(cancellationToken); + if (ch != bytes[0]) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, $"Unexpected character: {(char) ch}"); + } + } + + /// <summary> + /// Write the bytes in array buf as a JSON characters, escaping as needed + /// </summary> + private async Task WriteJsonStringAsync(byte[] bytes, CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + + var len = bytes.Length; + for (var i = 0; i < len; i++) + { + if ((bytes[i] & 0x00FF) >= 0x30) + { + if (bytes[i] == TJSONProtocolConstants.Backslash[0]) + { + await Trans.WriteAsync(TJSONProtocolConstants.Backslash, cancellationToken); + await Trans.WriteAsync(TJSONProtocolConstants.Backslash, cancellationToken); + } + else + { + await Trans.WriteAsync(bytes.ToArray(), i, 1, cancellationToken); + } + } + else + { + _tempBuffer[0] = TJSONProtocolConstants.JsonCharTable[bytes[i]]; + if (_tempBuffer[0] == 1) + { + await Trans.WriteAsync(bytes, i, 1, cancellationToken); + } + else if (_tempBuffer[0] > 1) + { + await Trans.WriteAsync(TJSONProtocolConstants.Backslash, cancellationToken); + await Trans.WriteAsync(_tempBuffer, 0, 1, cancellationToken); + } + else + { + await Trans.WriteAsync(TJSONProtocolConstants.EscSequences, cancellationToken); + _tempBuffer[0] = TJSONProtocolHelper.ToHexChar((byte) (bytes[i] >> 4)); + _tempBuffer[1] = TJSONProtocolHelper.ToHexChar(bytes[i]); + await Trans.WriteAsync(_tempBuffer, 0, 2, cancellationToken); + } + } + } + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + /// <summary> + /// Write out number as a JSON value. If the context dictates so, it will be + /// wrapped in quotes to output as a JSON string. + /// </summary> + private async Task WriteJsonIntegerAsync(long num, CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + var str = num.ToString(); + + var escapeNum = Context.EscapeNumbers(); + if (escapeNum) + { + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + var bytes = Utf8Encoding.GetBytes(str); + await Trans.WriteAsync(bytes, cancellationToken); + + if (escapeNum) + { + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + } + + /// <summary> + /// Write out a double as a JSON value. If it is NaN or infinity or if the + /// context dictates escaping, Write out as JSON string. + /// </summary> + private async Task WriteJsonDoubleAsync(double num, CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + var str = num.ToString("G17", CultureInfo.InvariantCulture); + var special = false; + + switch (str[0]) + { + case 'N': // NaN + case 'I': // Infinity + special = true; + break; + case '-': + if (str[1] == 'I') + { + // -Infinity + special = true; + } + break; + } + + var escapeNum = special || Context.EscapeNumbers(); + + if (escapeNum) + { + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + await Trans.WriteAsync(Utf8Encoding.GetBytes(str), cancellationToken); + + if (escapeNum) + { + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + } + + /// <summary> + /// Write out contents of byte array b as a JSON string with base-64 encoded + /// data + /// </summary> + private async Task WriteJsonBase64Async(byte[] bytes, CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + + var len = bytes.Length; + var off = 0; + + while (len >= 3) + { + // Encode 3 bytes at a time + TBase64Helper.Encode(bytes, off, 3, _tempBuffer, 0); + await Trans.WriteAsync(_tempBuffer, 0, 4, cancellationToken); + off += 3; + len -= 3; + } + + if (len > 0) + { + // Encode remainder + TBase64Helper.Encode(bytes, off, len, _tempBuffer, 0); + await Trans.WriteAsync(_tempBuffer, 0, len + 1, cancellationToken); + } + + await Trans.WriteAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + private async Task WriteJsonObjectStartAsync(CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + await Trans.WriteAsync(TJSONProtocolConstants.LeftBrace, cancellationToken); + PushContext(new JSONPairContext(this)); + } + + private async Task WriteJsonObjectEndAsync(CancellationToken cancellationToken) + { + PopContext(); + await Trans.WriteAsync(TJSONProtocolConstants.RightBrace, cancellationToken); + } + + private async Task WriteJsonArrayStartAsync(CancellationToken cancellationToken) + { + await Context.WriteAsync(cancellationToken); + await Trans.WriteAsync(TJSONProtocolConstants.LeftBracket, cancellationToken); + PushContext(new JSONListContext(this)); + } + + private async Task WriteJsonArrayEndAsync(CancellationToken cancellationToken) + { + PopContext(); + await Trans.WriteAsync(TJSONProtocolConstants.RightBracket, cancellationToken); + } + + public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken) + { + await WriteJsonArrayStartAsync(cancellationToken); + await WriteJsonIntegerAsync(Version, cancellationToken); + + var b = Utf8Encoding.GetBytes(message.Name); + await WriteJsonStringAsync(b, cancellationToken); + + await WriteJsonIntegerAsync((long) message.Type, cancellationToken); + await WriteJsonIntegerAsync(message.SeqID, cancellationToken); + } + + public override async Task WriteMessageEndAsync(CancellationToken cancellationToken) + { + await WriteJsonArrayEndAsync(cancellationToken); + } + + public override async Task WriteStructBeginAsync(TStruct @struct, CancellationToken cancellationToken) + { + await WriteJsonObjectStartAsync(cancellationToken); + } + + public override async Task WriteStructEndAsync(CancellationToken cancellationToken) + { + await WriteJsonObjectEndAsync(cancellationToken); + } + + public override async Task WriteFieldBeginAsync(TField field, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(field.ID, cancellationToken); + await WriteJsonObjectStartAsync(cancellationToken); + await WriteJsonStringAsync(TJSONProtocolHelper.GetTypeNameForTypeId(field.Type), cancellationToken); + } + + public override async Task WriteFieldEndAsync(CancellationToken cancellationToken) + { + await WriteJsonObjectEndAsync(cancellationToken); + } + + public override async Task WriteFieldStopAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override async Task WriteMapBeginAsync(TMap map, CancellationToken cancellationToken) + { + await WriteJsonArrayStartAsync(cancellationToken); + await WriteJsonStringAsync(TJSONProtocolHelper.GetTypeNameForTypeId(map.KeyType), cancellationToken); + await WriteJsonStringAsync(TJSONProtocolHelper.GetTypeNameForTypeId(map.ValueType), cancellationToken); + await WriteJsonIntegerAsync(map.Count, cancellationToken); + await WriteJsonObjectStartAsync(cancellationToken); + } + + public override async Task WriteMapEndAsync(CancellationToken cancellationToken) + { + await WriteJsonObjectEndAsync(cancellationToken); + await WriteJsonArrayEndAsync(cancellationToken); + } + + public override async Task WriteListBeginAsync(TList list, CancellationToken cancellationToken) + { + await WriteJsonArrayStartAsync(cancellationToken); + await WriteJsonStringAsync(TJSONProtocolHelper.GetTypeNameForTypeId(list.ElementType), cancellationToken); + await WriteJsonIntegerAsync(list.Count, cancellationToken); + } + + public override async Task WriteListEndAsync(CancellationToken cancellationToken) + { + await WriteJsonArrayEndAsync(cancellationToken); + } + + public override async Task WriteSetBeginAsync(TSet set, CancellationToken cancellationToken) + { + await WriteJsonArrayStartAsync(cancellationToken); + await WriteJsonStringAsync(TJSONProtocolHelper.GetTypeNameForTypeId(set.ElementType), cancellationToken); + await WriteJsonIntegerAsync(set.Count, cancellationToken); + } + + public override async Task WriteSetEndAsync(CancellationToken cancellationToken) + { + await WriteJsonArrayEndAsync(cancellationToken); + } + + public override async Task WriteBoolAsync(bool b, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(b ? 1 : 0, cancellationToken); + } + + public override async Task WriteByteAsync(sbyte b, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(b, cancellationToken); + } + + public override async Task WriteI16Async(short i16, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(i16, cancellationToken); + } + + public override async Task WriteI32Async(int i32, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(i32, cancellationToken); + } + + public override async Task WriteI64Async(long i64, CancellationToken cancellationToken) + { + await WriteJsonIntegerAsync(i64, cancellationToken); + } + + public override async Task WriteDoubleAsync(double d, CancellationToken cancellationToken) + { + await WriteJsonDoubleAsync(d, cancellationToken); + } + + public override async Task WriteStringAsync(string s, CancellationToken cancellationToken) + { + var b = Utf8Encoding.GetBytes(s); + await WriteJsonStringAsync(b, cancellationToken); + } + + public override async Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken) + { + await WriteJsonBase64Async(bytes, cancellationToken); + } + + /// <summary> + /// Read in a JSON string, unescaping as appropriate.. Skip Reading from the + /// context if skipContext is true. + /// </summary> + private async Task<byte[]> ReadJsonStringAsync(bool skipContext, CancellationToken cancellationToken) + { + using (var buffer = new MemoryStream()) + { + var codeunits = new List<char>(); + + + if (!skipContext) + { + await Context.ReadAsync(cancellationToken); + } + + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.Quote, cancellationToken); + + while (true) + { + var ch = await Reader.ReadAsync(cancellationToken); + if (ch == TJSONProtocolConstants.Quote[0]) + { + break; + } + + // escaped? + if (ch != TJSONProtocolConstants.EscSequences[0]) + { + await buffer.WriteAsync(new[] {ch}, 0, 1, cancellationToken); + continue; + } + + // distinguish between \uXXXX and \? + ch = await Reader.ReadAsync(cancellationToken); + if (ch != TJSONProtocolConstants.EscSequences[1]) // control chars like \n + { + var off = Array.IndexOf(TJSONProtocolConstants.EscapeChars, (char) ch); + if (off == -1) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected control char"); + } + ch = TJSONProtocolConstants.EscapeCharValues[off]; + await buffer.WriteAsync(new[] {ch}, 0, 1, cancellationToken); + continue; + } + + // it's \uXXXX + await Trans.ReadAllAsync(_tempBuffer, 0, 4, cancellationToken); + + var wch = (short) ((TJSONProtocolHelper.ToHexVal(_tempBuffer[0]) << 12) + + (TJSONProtocolHelper.ToHexVal(_tempBuffer[1]) << 8) + + (TJSONProtocolHelper.ToHexVal(_tempBuffer[2]) << 4) + + TJSONProtocolHelper.ToHexVal(_tempBuffer[3])); + + if (char.IsHighSurrogate((char) wch)) + { + if (codeunits.Count > 0) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected low surrogate char"); + } + codeunits.Add((char) wch); + } + else if (char.IsLowSurrogate((char) wch)) + { + if (codeunits.Count == 0) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected high surrogate char"); + } + + codeunits.Add((char) wch); + var tmp = Utf8Encoding.GetBytes(codeunits.ToArray()); + await buffer.WriteAsync(tmp, 0, tmp.Length, cancellationToken); + codeunits.Clear(); + } + else + { + var tmp = Utf8Encoding.GetBytes(new[] {(char) wch}); + await buffer.WriteAsync(tmp, 0, tmp.Length, cancellationToken); + } + } + + if (codeunits.Count > 0) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected low surrogate char"); + } + + return buffer.ToArray(); + } + } + + /// <summary> + /// Read in a sequence of characters that are all valid in JSON numbers. Does + /// not do a complete regex check to validate that this is actually a number. + /// </summary> + private async Task<string> ReadJsonNumericCharsAsync(CancellationToken cancellationToken) + { + var strbld = new StringBuilder(); + while (true) + { + //TODO: workaround for primitive types with TJsonProtocol, think - how to rewrite into more easy form without exceptions + try + { + var ch = await Reader.PeekAsync(cancellationToken); + if (!TJSONProtocolHelper.IsJsonNumeric(ch)) + { + break; + } + var c = (char)await Reader.ReadAsync(cancellationToken); + strbld.Append(c); + } + catch (TTransportException) + { + break; + } + } + return strbld.ToString(); + } + + /// <summary> + /// Read in a JSON number. If the context dictates, Read in enclosing quotes. + /// </summary> + private async Task<long> ReadJsonIntegerAsync(CancellationToken cancellationToken) + { + await Context.ReadAsync(cancellationToken); + if (Context.EscapeNumbers()) + { + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + var str = await ReadJsonNumericCharsAsync(cancellationToken); + if (Context.EscapeNumbers()) + { + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + try + { + return long.Parse(str); + } + catch (FormatException) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Bad data encounted in numeric data"); + } + } + + /// <summary> + /// Read in a JSON double value. Throw if the value is not wrapped in quotes + /// when expected or if wrapped in quotes when not expected. + /// </summary> + private async Task<double> ReadJsonDoubleAsync(CancellationToken cancellationToken) + { + await Context.ReadAsync(cancellationToken); + if (await Reader.PeekAsync(cancellationToken) == TJSONProtocolConstants.Quote[0]) + { + var arr = await ReadJsonStringAsync(true, cancellationToken); + var dub = double.Parse(Utf8Encoding.GetString(arr, 0, arr.Length), CultureInfo.InvariantCulture); + + if (!Context.EscapeNumbers() && !double.IsNaN(dub) && !double.IsInfinity(dub)) + { + // Throw exception -- we should not be in a string in this case + throw new TProtocolException(TProtocolException.INVALID_DATA, "Numeric data unexpectedly quoted"); + } + + return dub; + } + + if (Context.EscapeNumbers()) + { + // This will throw - we should have had a quote if escapeNum == true + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.Quote, cancellationToken); + } + + try + { + return double.Parse(await ReadJsonNumericCharsAsync(cancellationToken), CultureInfo.InvariantCulture); + } + catch (FormatException) + { + throw new TProtocolException(TProtocolException.INVALID_DATA, "Bad data encounted in numeric data"); + } + } + + /// <summary> + /// Read in a JSON string containing base-64 encoded data and decode it. + /// </summary> + private async Task<byte[]> ReadJsonBase64Async(CancellationToken cancellationToken) + { + var b = await ReadJsonStringAsync(false, cancellationToken); + var len = b.Length; + var off = 0; + var size = 0; + + // reduce len to ignore fill bytes + while ((len > 0) && (b[len - 1] == '=')) + { + --len; + } + + // read & decode full byte triplets = 4 source bytes + while (len > 4) + { + // Decode 4 bytes at a time + TBase64Helper.Decode(b, off, 4, b, size); // NB: decoded in place + off += 4; + len -= 4; + size += 3; + } + + // Don't decode if we hit the end or got a single leftover byte (invalid + // base64 but legal for skip of regular string exType) + if (len > 1) + { + // Decode remainder + TBase64Helper.Decode(b, off, len, b, size); // NB: decoded in place + size += len - 1; + } + + // Sadly we must copy the byte[] (any way around this?) + var result = new byte[size]; + Array.Copy(b, 0, result, 0, size); + return result; + } + + private async Task ReadJsonObjectStartAsync(CancellationToken cancellationToken) + { + await Context.ReadAsync(cancellationToken); + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.LeftBrace, cancellationToken); + PushContext(new JSONPairContext(this)); + } + + private async Task ReadJsonObjectEndAsync(CancellationToken cancellationToken) + { + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.RightBrace, cancellationToken); + PopContext(); + } + + private async Task ReadJsonArrayStartAsync(CancellationToken cancellationToken) + { + await Context.ReadAsync(cancellationToken); + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.LeftBracket, cancellationToken); + PushContext(new JSONListContext(this)); + } + + private async Task ReadJsonArrayEndAsync(CancellationToken cancellationToken) + { + await ReadJsonSyntaxCharAsync(TJSONProtocolConstants.RightBracket, cancellationToken); + PopContext(); + } + + public override async Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken) + { + var message = new TMessage(); + await ReadJsonArrayStartAsync(cancellationToken); + if (await ReadJsonIntegerAsync(cancellationToken) != Version) + { + throw new TProtocolException(TProtocolException.BAD_VERSION, "Message contained bad version."); + } + + var buf = await ReadJsonStringAsync(false, cancellationToken); + message.Name = Utf8Encoding.GetString(buf, 0, buf.Length); + message.Type = (TMessageType) await ReadJsonIntegerAsync(cancellationToken); + message.SeqID = (int) await ReadJsonIntegerAsync(cancellationToken); + return message; + } + + public override async Task ReadMessageEndAsync(CancellationToken cancellationToken) + { + await ReadJsonArrayEndAsync(cancellationToken); + } + + public override async Task<TStruct> ReadStructBeginAsync(CancellationToken cancellationToken) + { + await ReadJsonObjectStartAsync(cancellationToken); + return new TStruct(); + } + + public override async Task ReadStructEndAsync(CancellationToken cancellationToken) + { + await ReadJsonObjectEndAsync(cancellationToken); + } + + public override async Task<TField> ReadFieldBeginAsync(CancellationToken cancellationToken) + { + var field = new TField(); + var ch = await Reader.PeekAsync(cancellationToken); + if (ch == TJSONProtocolConstants.RightBrace[0]) + { + field.Type = TType.Stop; + } + else + { + field.ID = (short) await ReadJsonIntegerAsync(cancellationToken); + await ReadJsonObjectStartAsync(cancellationToken); + field.Type = TJSONProtocolHelper.GetTypeIdForTypeName(await ReadJsonStringAsync(false, cancellationToken)); + } + return field; + } + + public override async Task ReadFieldEndAsync(CancellationToken cancellationToken) + { + await ReadJsonObjectEndAsync(cancellationToken); + } + + public override async Task<TMap> ReadMapBeginAsync(CancellationToken cancellationToken) + { + var map = new TMap(); + await ReadJsonArrayStartAsync(cancellationToken); + map.KeyType = TJSONProtocolHelper.GetTypeIdForTypeName(await ReadJsonStringAsync(false, cancellationToken)); + map.ValueType = TJSONProtocolHelper.GetTypeIdForTypeName(await ReadJsonStringAsync(false, cancellationToken)); + map.Count = (int) await ReadJsonIntegerAsync(cancellationToken); + await ReadJsonObjectStartAsync(cancellationToken); + return map; + } + + public override async Task ReadMapEndAsync(CancellationToken cancellationToken) + { + await ReadJsonObjectEndAsync(cancellationToken); + await ReadJsonArrayEndAsync(cancellationToken); + } + + public override async Task<TList> ReadListBeginAsync(CancellationToken cancellationToken) + { + var list = new TList(); + await ReadJsonArrayStartAsync(cancellationToken); + list.ElementType = TJSONProtocolHelper.GetTypeIdForTypeName(await ReadJsonStringAsync(false, cancellationToken)); + list.Count = (int) await ReadJsonIntegerAsync(cancellationToken); + return list; + } + + public override async Task ReadListEndAsync(CancellationToken cancellationToken) + { + await ReadJsonArrayEndAsync(cancellationToken); + } + + public override async Task<TSet> ReadSetBeginAsync(CancellationToken cancellationToken) + { + var set = new TSet(); + await ReadJsonArrayStartAsync(cancellationToken); + set.ElementType = TJSONProtocolHelper.GetTypeIdForTypeName(await ReadJsonStringAsync(false, cancellationToken)); + set.Count = (int) await ReadJsonIntegerAsync(cancellationToken); + return set; + } + + public override async Task ReadSetEndAsync(CancellationToken cancellationToken) + { + await ReadJsonArrayEndAsync(cancellationToken); + } + + public override async Task<bool> ReadBoolAsync(CancellationToken cancellationToken) + { + return await ReadJsonIntegerAsync(cancellationToken) != 0; + } + + public override async Task<sbyte> ReadByteAsync(CancellationToken cancellationToken) + { + return (sbyte) await ReadJsonIntegerAsync(cancellationToken); + } + + public override async Task<short> ReadI16Async(CancellationToken cancellationToken) + { + return (short) await ReadJsonIntegerAsync(cancellationToken); + } + + public override async Task<int> ReadI32Async(CancellationToken cancellationToken) + { + return (int) await ReadJsonIntegerAsync(cancellationToken); + } + + public override async Task<long> ReadI64Async(CancellationToken cancellationToken) + { + return await ReadJsonIntegerAsync(cancellationToken); + } + + public override async Task<double> ReadDoubleAsync(CancellationToken cancellationToken) + { + return await ReadJsonDoubleAsync(cancellationToken); + } + + public override async Task<string> ReadStringAsync(CancellationToken cancellationToken) + { + var buf = await ReadJsonStringAsync(false, cancellationToken); + return Utf8Encoding.GetString(buf, 0, buf.Length); + } + + public override async Task<byte[]> ReadBinaryAsync(CancellationToken cancellationToken) + { + return await ReadJsonBase64Async(cancellationToken); + } + + /// <summary> + /// Factory for JSON protocol objects + /// </summary> + public class Factory : ITProtocolFactory + { + public TProtocol GetProtocol(TClientTransport trans) + { + return new TJsonProtocol(trans); + } + } + + /// <summary> + /// Base class for tracking JSON contexts that may require + /// inserting/Reading additional JSON syntax characters + /// This base context does nothing. + /// </summary> + protected class JSONBaseContext + { + protected TJsonProtocol Proto; + + public JSONBaseContext(TJsonProtocol proto) + { + Proto = proto; + } + + public virtual async Task WriteAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public virtual async Task ReadAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public virtual bool EscapeNumbers() + { + return false; + } + } + + /// <summary> + /// Context for JSON lists. Will insert/Read commas before each item except + /// for the first one + /// </summary> + protected class JSONListContext : JSONBaseContext + { + private bool _first = true; + + public JSONListContext(TJsonProtocol protocol) + : base(protocol) + { + } + + public override async Task WriteAsync(CancellationToken cancellationToken) + { + if (_first) + { + _first = false; + } + else + { + await Proto.Trans.WriteAsync(TJSONProtocolConstants.Comma, cancellationToken); + } + } + + public override async Task ReadAsync(CancellationToken cancellationToken) + { + if (_first) + { + _first = false; + } + else + { + await Proto.ReadJsonSyntaxCharAsync(TJSONProtocolConstants.Comma, cancellationToken); + } + } + } + + /// <summary> + /// Context for JSON records. Will insert/Read colons before the value portion + /// of each record pair, and commas before each key except the first. In + /// addition, will indicate that numbers in the key position need to be + /// escaped in quotes (since JSON keys must be strings). + /// </summary> + // ReSharper disable once InconsistentNaming + protected class JSONPairContext : JSONBaseContext + { + private bool _colon = true; + + private bool _first = true; + + public JSONPairContext(TJsonProtocol proto) + : base(proto) + { + } + + public override async Task WriteAsync(CancellationToken cancellationToken) + { + if (_first) + { + _first = false; + _colon = true; + } + else + { + await Proto.Trans.WriteAsync(_colon ? TJSONProtocolConstants.Colon : TJSONProtocolConstants.Comma, cancellationToken); + _colon = !_colon; + } + } + + public override async Task ReadAsync(CancellationToken cancellationToken) + { + if (_first) + { + _first = false; + _colon = true; + } + else + { + await Proto.ReadJsonSyntaxCharAsync(_colon ? TJSONProtocolConstants.Colon : TJSONProtocolConstants.Comma, cancellationToken); + _colon = !_colon; + } + } + + public override bool EscapeNumbers() + { + return _colon; + } + } + + /// <summary> + /// Holds up to one byte from the transport + /// </summary> + protected class LookaheadReader + { + private readonly byte[] _data = new byte[1]; + + private bool _hasData; + protected TJsonProtocol Proto; + + public LookaheadReader(TJsonProtocol proto) + { + Proto = proto; + } + + /// <summary> + /// Return and consume the next byte to be Read, either taking it from the + /// data buffer if present or getting it from the transport otherwise. + /// </summary> + public async Task<byte> ReadAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<byte>(cancellationToken); + } + + if (_hasData) + { + _hasData = false; + } + else + { + // find more easy way to avoid exception on reading primitive types + await Proto.Trans.ReadAllAsync(_data, 0, 1, cancellationToken); + } + return _data[0]; + } + + /// <summary> + /// Return the next byte to be Read without consuming, filling the data + /// buffer if it has not been filled alReady. + /// </summary> + public async Task<byte> PeekAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<byte>(cancellationToken); + } + + if (!_hasData) + { + // find more easy way to avoid exception on reading primitive types + await Proto.Trans.ReadAllAsync(_data, 0, 1, cancellationToken); + } + _hasData = true; + return _data[0]; + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TMultiplexedProtocol.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TMultiplexedProtocol.cs new file mode 100644 index 000000000..367e4e644 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TMultiplexedProtocol.cs @@ -0,0 +1,91 @@ +// 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; +using Thrift.Protocols.Entities; + +namespace Thrift.Protocols +{ + /** + * TMultiplexedProtocol is a protocol-independent concrete decorator that allows a Thrift + * client to communicate with a multiplexing Thrift server, by prepending the service name + * to the function name during function calls. + * + * NOTE: THIS IS NOT TO BE USED BY SERVERS. + * On the server, use TMultiplexedProcessor to handle requests from a multiplexing client. + * + * This example uses a single socket transport to invoke two services: + * + * TSocketClientTransport transport = new TSocketClientTransport("localhost", 9090); + * transport.open(); + * + * TBinaryProtocol protocol = new TBinaryProtocol(transport); + * + * TMultiplexedProtocol mp = new TMultiplexedProtocol(protocol, "Calculator"); + * Calculator.Client service = new Calculator.Client(mp); + * + * TMultiplexedProtocol mp2 = new TMultiplexedProtocol(protocol, "WeatherReport"); + * WeatherReport.Client service2 = new WeatherReport.Client(mp2); + * + * System.out.println(service.add(2,2)); + * System.out.println(service2.getTemperature()); + * + */ + + //TODO: implementation of TProtocol + + // ReSharper disable once InconsistentNaming + public class TMultiplexedProtocol : TProtocolDecorator + { + /** Used to delimit the service name from the function name */ + public const string Separator = ":"; + + private readonly string _serviceName; + + /** + * Wrap the specified protocol, allowing it to be used to communicate with a + * multiplexing server. The <code>serviceName</code> is required as it is + * prepended to the message header so that the multiplexing server can broker + * the function call to the proper service. + * + * Args: + * protocol Your communication protocol of choice, e.g. TBinaryProtocol + * serviceName The service name of the service communicating via this protocol. + */ + + public TMultiplexedProtocol(TProtocol protocol, string serviceName) + : base(protocol) + { + _serviceName = serviceName; + } + + public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken) + { + switch (message.Type) + { + case TMessageType.Call: + case TMessageType.Oneway: + await base.WriteMessageBeginAsync(new TMessage($"{_serviceName}{Separator}{message.Name}", message.Type, message.SeqID), cancellationToken); + break; + default: + await base.WriteMessageBeginAsync(message, cancellationToken); + break; + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocol.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocol.cs new file mode 100644 index 000000000..91e009d63 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocol.cs @@ -0,0 +1,376 @@ +// 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 Thrift.Protocols.Entities; +using Thrift.Transports; + +namespace Thrift.Protocols +{ + // ReSharper disable once InconsistentNaming + public abstract class TProtocol : IDisposable + { + public const int DefaultRecursionDepth = 64; + private bool _isDisposed; + protected int RecursionDepth; + + protected TClientTransport Trans; + + protected TProtocol(TClientTransport trans) + { + Trans = trans; + RecursionLimit = DefaultRecursionDepth; + RecursionDepth = 0; + } + + public TClientTransport Transport => Trans; + + protected int RecursionLimit { get; set; } + + public void Dispose() + { + Dispose(true); + } + + public void IncrementRecursionDepth() + { + if (RecursionDepth < RecursionLimit) + { + ++RecursionDepth; + } + else + { + throw new TProtocolException(TProtocolException.DEPTH_LIMIT, "Depth limit exceeded"); + } + } + + public void DecrementRecursionDepth() + { + --RecursionDepth; + } + + protected virtual void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + (Trans as IDisposable)?.Dispose(); + } + } + _isDisposed = true; + } + + public virtual async Task WriteMessageBeginAsync(TMessage message) + { + await WriteMessageBeginAsync(message, CancellationToken.None); + } + + public abstract Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken); + + public virtual async Task WriteMessageEndAsync() + { + await WriteMessageEndAsync(CancellationToken.None); + } + + public abstract Task WriteMessageEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteStructBeginAsync(TStruct @struct) + { + await WriteStructBeginAsync(@struct, CancellationToken.None); + } + + public abstract Task WriteStructBeginAsync(TStruct @struct, CancellationToken cancellationToken); + + public virtual async Task WriteStructEndAsync() + { + await WriteStructEndAsync(CancellationToken.None); + } + + public abstract Task WriteStructEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteFieldBeginAsync(TField field) + { + await WriteFieldBeginAsync(field, CancellationToken.None); + } + + public abstract Task WriteFieldBeginAsync(TField field, CancellationToken cancellationToken); + + public virtual async Task WriteFieldEndAsync() + { + await WriteFieldEndAsync(CancellationToken.None); + } + + public abstract Task WriteFieldEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteFieldStopAsync() + { + await WriteFieldStopAsync(CancellationToken.None); + } + + public abstract Task WriteFieldStopAsync(CancellationToken cancellationToken); + + public virtual async Task WriteMapBeginAsync(TMap map) + { + await WriteMapBeginAsync(map, CancellationToken.None); + } + + public abstract Task WriteMapBeginAsync(TMap map, CancellationToken cancellationToken); + + public virtual async Task WriteMapEndAsync() + { + await WriteMapEndAsync(CancellationToken.None); + } + + public abstract Task WriteMapEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteListBeginAsync(TList list) + { + await WriteListBeginAsync(list, CancellationToken.None); + } + + public abstract Task WriteListBeginAsync(TList list, CancellationToken cancellationToken); + + public virtual async Task WriteListEndAsync() + { + await WriteListEndAsync(CancellationToken.None); + } + + public abstract Task WriteListEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteSetBeginAsync(TSet set) + { + await WriteSetBeginAsync(set, CancellationToken.None); + } + + public abstract Task WriteSetBeginAsync(TSet set, CancellationToken cancellationToken); + + public virtual async Task WriteSetEndAsync() + { + await WriteSetEndAsync(CancellationToken.None); + } + + public abstract Task WriteSetEndAsync(CancellationToken cancellationToken); + + public virtual async Task WriteBoolAsync(bool b) + { + await WriteBoolAsync(b, CancellationToken.None); + } + + public abstract Task WriteBoolAsync(bool b, CancellationToken cancellationToken); + + public virtual async Task WriteByteAsync(sbyte b) + { + await WriteByteAsync(b, CancellationToken.None); + } + + public abstract Task WriteByteAsync(sbyte b, CancellationToken cancellationToken); + + public virtual async Task WriteI16Async(short i16) + { + await WriteI16Async(i16, CancellationToken.None); + } + + public abstract Task WriteI16Async(short i16, CancellationToken cancellationToken); + + public virtual async Task WriteI32Async(int i32) + { + await WriteI32Async(i32, CancellationToken.None); + } + + public abstract Task WriteI32Async(int i32, CancellationToken cancellationToken); + + public virtual async Task WriteI64Async(long i64) + { + await WriteI64Async(i64, CancellationToken.None); + } + + public abstract Task WriteI64Async(long i64, CancellationToken cancellationToken); + + public virtual async Task WriteDoubleAsync(double d) + { + await WriteDoubleAsync(d, CancellationToken.None); + } + + public abstract Task WriteDoubleAsync(double d, CancellationToken cancellationToken); + + public virtual async Task WriteStringAsync(string s) + { + await WriteStringAsync(s, CancellationToken.None); + } + + public virtual async Task WriteStringAsync(string s, CancellationToken cancellationToken) + { + var bytes = Encoding.UTF8.GetBytes(s); + await WriteBinaryAsync(bytes, cancellationToken); + } + + public virtual async Task WriteBinaryAsync(byte[] bytes) + { + await WriteBinaryAsync(bytes, CancellationToken.None); + } + + public abstract Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken); + + public virtual async Task<TMessage> ReadMessageBeginAsync() + { + return await ReadMessageBeginAsync(CancellationToken.None); + } + + public abstract Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadMessageEndAsync() + { + await ReadMessageEndAsync(CancellationToken.None); + } + + public abstract Task ReadMessageEndAsync(CancellationToken cancellationToken); + + public virtual async Task<TStruct> ReadStructBeginAsync() + { + return await ReadStructBeginAsync(CancellationToken.None); + } + + public abstract Task<TStruct> ReadStructBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadStructEndAsync() + { + await ReadStructEndAsync(CancellationToken.None); + } + + public abstract Task ReadStructEndAsync(CancellationToken cancellationToken); + + public virtual async Task<TField> ReadFieldBeginAsync() + { + return await ReadFieldBeginAsync(CancellationToken.None); + } + + public abstract Task<TField> ReadFieldBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadFieldEndAsync() + { + await ReadFieldEndAsync(CancellationToken.None); + } + + public abstract Task ReadFieldEndAsync(CancellationToken cancellationToken); + + public virtual async Task<TMap> ReadMapBeginAsync() + { + return await ReadMapBeginAsync(CancellationToken.None); + } + + public abstract Task<TMap> ReadMapBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadMapEndAsync() + { + await ReadMapEndAsync(CancellationToken.None); + } + + public abstract Task ReadMapEndAsync(CancellationToken cancellationToken); + + public virtual async Task<TList> ReadListBeginAsync() + { + return await ReadListBeginAsync(CancellationToken.None); + } + + public abstract Task<TList> ReadListBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadListEndAsync() + { + await ReadListEndAsync(CancellationToken.None); + } + + public abstract Task ReadListEndAsync(CancellationToken cancellationToken); + + public virtual async Task<TSet> ReadSetBeginAsync() + { + return await ReadSetBeginAsync(CancellationToken.None); + } + + public abstract Task<TSet> ReadSetBeginAsync(CancellationToken cancellationToken); + + public virtual async Task ReadSetEndAsync() + { + await ReadSetEndAsync(CancellationToken.None); + } + + public abstract Task ReadSetEndAsync(CancellationToken cancellationToken); + + public virtual async Task<bool> ReadBoolAsync() + { + return await ReadBoolAsync(CancellationToken.None); + } + + public abstract Task<bool> ReadBoolAsync(CancellationToken cancellationToken); + + public virtual async Task<sbyte> ReadByteAsync() + { + return await ReadByteAsync(CancellationToken.None); + } + + public abstract Task<sbyte> ReadByteAsync(CancellationToken cancellationToken); + + public virtual async Task<short> ReadI16Async() + { + return await ReadI16Async(CancellationToken.None); + } + + public abstract Task<short> ReadI16Async(CancellationToken cancellationToken); + + public virtual async Task<int> ReadI32Async() + { + return await ReadI32Async(CancellationToken.None); + } + + public abstract Task<int> ReadI32Async(CancellationToken cancellationToken); + + public virtual async Task<long> ReadI64Async() + { + return await ReadI64Async(CancellationToken.None); + } + + public abstract Task<long> ReadI64Async(CancellationToken cancellationToken); + + public virtual async Task<double> ReadDoubleAsync() + { + return await ReadDoubleAsync(CancellationToken.None); + } + + public abstract Task<double> ReadDoubleAsync(CancellationToken cancellationToken); + + public virtual async Task<string> ReadStringAsync() + { + return await ReadStringAsync(CancellationToken.None); + } + + public virtual async Task<string> ReadStringAsync(CancellationToken cancellationToken) + { + var buf = await ReadBinaryAsync(cancellationToken); + return Encoding.UTF8.GetString(buf, 0, buf.Length); + } + + public virtual async Task<byte[]> ReadBinaryAsync() + { + return await ReadBinaryAsync(CancellationToken.None); + } + + public abstract Task<byte[]> ReadBinaryAsync(CancellationToken cancellationToken); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolDecorator.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolDecorator.cs new file mode 100644 index 000000000..3222754a8 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolDecorator.cs @@ -0,0 +1,247 @@ +// 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.Threading; +using System.Threading.Tasks; +using Thrift.Protocols.Entities; + +namespace Thrift.Protocols +{ + // ReSharper disable once InconsistentNaming + /// <summary> + /// TProtocolDecorator forwards all requests to an enclosed TProtocol instance, + /// providing a way to author concise concrete decorator subclasses.While it has + /// no abstract methods, it is marked abstract as a reminder that by itself, + /// it does not modify the behaviour of the enclosed TProtocol. + /// </summary> + public abstract class TProtocolDecorator : TProtocol + { + private readonly TProtocol _wrappedProtocol; + + protected TProtocolDecorator(TProtocol protocol) + : base(protocol.Transport) + { + _wrappedProtocol = protocol ?? throw new ArgumentNullException(nameof(protocol)); + } + + public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteMessageBeginAsync(message, cancellationToken); + } + + public override async Task WriteMessageEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteMessageEndAsync(cancellationToken); + } + + public override async Task WriteStructBeginAsync(TStruct @struct, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteStructBeginAsync(@struct, cancellationToken); + } + + public override async Task WriteStructEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteStructEndAsync(cancellationToken); + } + + public override async Task WriteFieldBeginAsync(TField field, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteFieldBeginAsync(field, cancellationToken); + } + + public override async Task WriteFieldEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteFieldEndAsync(cancellationToken); + } + + public override async Task WriteFieldStopAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteFieldStopAsync(cancellationToken); + } + + public override async Task WriteMapBeginAsync(TMap map, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteMapBeginAsync(map, cancellationToken); + } + + public override async Task WriteMapEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteMapEndAsync(cancellationToken); + } + + public override async Task WriteListBeginAsync(TList list, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteListBeginAsync(list, cancellationToken); + } + + public override async Task WriteListEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteListEndAsync(cancellationToken); + } + + public override async Task WriteSetBeginAsync(TSet set, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteSetBeginAsync(set, cancellationToken); + } + + public override async Task WriteSetEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteSetEndAsync(cancellationToken); + } + + public override async Task WriteBoolAsync(bool b, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteBoolAsync(b, cancellationToken); + } + + public override async Task WriteByteAsync(sbyte b, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteByteAsync(b, cancellationToken); + } + + public override async Task WriteI16Async(short i16, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteI16Async(i16, cancellationToken); + } + + public override async Task WriteI32Async(int i32, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteI32Async(i32, cancellationToken); + } + + public override async Task WriteI64Async(long i64, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteI64Async(i64, cancellationToken); + } + + public override async Task WriteDoubleAsync(double d, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteDoubleAsync(d, cancellationToken); + } + + public override async Task WriteStringAsync(string s, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteStringAsync(s, cancellationToken); + } + + public override async Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken) + { + await _wrappedProtocol.WriteBinaryAsync(bytes, cancellationToken); + } + + public override async Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadMessageBeginAsync(cancellationToken); + } + + public override async Task ReadMessageEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadMessageEndAsync(cancellationToken); + } + + public override async Task<TStruct> ReadStructBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadStructBeginAsync(cancellationToken); + } + + public override async Task ReadStructEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadStructEndAsync(cancellationToken); + } + + public override async Task<TField> ReadFieldBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadFieldBeginAsync(cancellationToken); + } + + public override async Task ReadFieldEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadFieldEndAsync(cancellationToken); + } + + public override async Task<TMap> ReadMapBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadMapBeginAsync(cancellationToken); + } + + public override async Task ReadMapEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadMapEndAsync(cancellationToken); + } + + public override async Task<TList> ReadListBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadListBeginAsync(cancellationToken); + } + + public override async Task ReadListEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadListEndAsync(cancellationToken); + } + + public override async Task<TSet> ReadSetBeginAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadSetBeginAsync(cancellationToken); + } + + public override async Task ReadSetEndAsync(CancellationToken cancellationToken) + { + await _wrappedProtocol.ReadSetEndAsync(cancellationToken); + } + + public override async Task<bool> ReadBoolAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadBoolAsync(cancellationToken); + } + + public override async Task<sbyte> ReadByteAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadByteAsync(cancellationToken); + } + + public override async Task<short> ReadI16Async(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadI16Async(cancellationToken); + } + + public override async Task<int> ReadI32Async(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadI32Async(cancellationToken); + } + + public override async Task<long> ReadI64Async(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadI64Async(cancellationToken); + } + + public override async Task<double> ReadDoubleAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadDoubleAsync(cancellationToken); + } + + public override async Task<string> ReadStringAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadStringAsync(cancellationToken); + } + + public override async Task<byte[]> ReadBinaryAsync(CancellationToken cancellationToken) + { + return await _wrappedProtocol.ReadBinaryAsync(cancellationToken); + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolException.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolException.cs new file mode 100644 index 000000000..8c67c3bfd --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/TProtocolException.cs @@ -0,0 +1,59 @@ +// 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. + +// ReSharper disable InconsistentNaming +namespace Thrift.Protocols +{ + public class TProtocolException : TException + { + // do not rename public constants - they used in generated files + public const int UNKNOWN = 0; + public const int INVALID_DATA = 1; + public const int NEGATIVE_SIZE = 2; + public const int SIZE_LIMIT = 3; + public const int BAD_VERSION = 4; + public const int NOT_IMPLEMENTED = 5; + public const int DEPTH_LIMIT = 6; + + protected int Type = UNKNOWN; + + public TProtocolException() + { + } + + public TProtocolException(int type) + { + Type = type; + } + + public TProtocolException(int type, string message) + : base(message) + { + Type = type; + } + + public TProtocolException(string message) + : base(message) + { + } + + public int GetExceptionType() + { + return Type; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Helper.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Helper.cs new file mode 100644 index 000000000..7eff5e181 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Helper.cs @@ -0,0 +1,101 @@ +// 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.Protocols.Utilities +{ + // ReSharper disable once InconsistentNaming + internal static class TBase64Helper + { + //TODO: Constants + //TODO: Check for args + //TODO: Unitests + + internal const string EncodeTable = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + + private static readonly int[] DecodeTable = + { + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 62, -1, -1, -1, 63, + 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, -1, -1, -1, -1, -1, -1, + -1, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, + 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, -1, -1, -1, -1, -1, + -1, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, + 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1 + }; + + internal static void Encode(byte[] src, int srcOff, int len, byte[] dst, int dstOff) + { + if (src == null) + { + throw new ArgumentNullException(nameof(src)); + } + + dst[dstOff] = (byte) EncodeTable[(src[srcOff] >> 2) & 0x3F]; + + if (len == 3) + { + dst[dstOff + 1] = (byte) EncodeTable[((src[srcOff] << 4) & 0x30) | ((src[srcOff + 1] >> 4) & 0x0F)]; + dst[dstOff + 2] = (byte) EncodeTable[((src[srcOff + 1] << 2) & 0x3C) | ((src[srcOff + 2] >> 6) & 0x03)]; + dst[dstOff + 3] = (byte) EncodeTable[src[srcOff + 2] & 0x3F]; + } + else if (len == 2) + { + dst[dstOff + 1] = (byte) EncodeTable[((src[srcOff] << 4) & 0x30) | ((src[srcOff + 1] >> 4) & 0x0F)]; + dst[dstOff + 2] = (byte) EncodeTable[(src[srcOff + 1] << 2) & 0x3C]; + } + else + { + // len == 1 + dst[dstOff + 1] = (byte) EncodeTable[(src[srcOff] << 4) & 0x30]; + } + } + + internal static void Decode(byte[] src, int srcOff, int len, byte[] dst, int dstOff) + { + if (src == null) + { + throw new ArgumentNullException(nameof(src)); + } + + dst[dstOff] = (byte) ((DecodeTable[src[srcOff] & 0x0FF] << 2) | (DecodeTable[src[srcOff + 1] & 0x0FF] >> 4)); + + if (len > 2) + { + dst[dstOff + 1] = + (byte) + (((DecodeTable[src[srcOff + 1] & 0x0FF] << 4) & 0xF0) | (DecodeTable[src[srcOff + 2] & 0x0FF] >> 2)); + if (len > 3) + { + dst[dstOff + 2] = + (byte) + (((DecodeTable[src[srcOff + 2] & 0x0FF] << 6) & 0xC0) | DecodeTable[src[srcOff + 3] & 0x0FF]); + } + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Utils.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Utils.cs new file mode 100644 index 000000000..15fd45cbe --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TBase64Utils.cs @@ -0,0 +1,101 @@ +// 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.Protocols.Utilities +{ + // ReSharper disable once InconsistentNaming + internal static class TBase64Utils + { + //TODO: Constants + //TODO: Check for args + //TODO: Unitests + + internal const string EncodeTable = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + + private static readonly int[] DecodeTable = + { + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 62, -1, -1, -1, 63, + 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, -1, -1, -1, -1, -1, -1, + -1, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, + 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, -1, -1, -1, -1, -1, + -1, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, + 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1 + }; + + internal static void Encode(byte[] src, int srcOff, int len, byte[] dst, int dstOff) + { + if (src == null) + { + throw new ArgumentNullException(nameof(src)); + } + + dst[dstOff] = (byte) EncodeTable[(src[srcOff] >> 2) & 0x3F]; + + if (len == 3) + { + dst[dstOff + 1] = (byte) EncodeTable[((src[srcOff] << 4) & 0x30) | ((src[srcOff + 1] >> 4) & 0x0F)]; + dst[dstOff + 2] = (byte) EncodeTable[((src[srcOff + 1] << 2) & 0x3C) | ((src[srcOff + 2] >> 6) & 0x03)]; + dst[dstOff + 3] = (byte) EncodeTable[src[srcOff + 2] & 0x3F]; + } + else if (len == 2) + { + dst[dstOff + 1] = (byte) EncodeTable[((src[srcOff] << 4) & 0x30) | ((src[srcOff + 1] >> 4) & 0x0F)]; + dst[dstOff + 2] = (byte) EncodeTable[(src[srcOff + 1] << 2) & 0x3C]; + } + else + { + // len == 1 + dst[dstOff + 1] = (byte) EncodeTable[(src[srcOff] << 4) & 0x30]; + } + } + + internal static void Decode(byte[] src, int srcOff, int len, byte[] dst, int dstOff) + { + if (src == null) + { + throw new ArgumentNullException(nameof(src)); + } + + dst[dstOff] = (byte) ((DecodeTable[src[srcOff] & 0x0FF] << 2) | (DecodeTable[src[srcOff + 1] & 0x0FF] >> 4)); + + if (len > 2) + { + dst[dstOff + 1] = + (byte) + (((DecodeTable[src[srcOff + 1] & 0x0FF] << 4) & 0xF0) | (DecodeTable[src[srcOff + 2] & 0x0FF] >> 2)); + if (len > 3) + { + dst[dstOff + 2] = + (byte) + (((DecodeTable[src[srcOff + 2] & 0x0FF] << 6) & 0xC0) | DecodeTable[src[srcOff + 3] & 0x0FF]); + } + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolConstants.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolConstants.cs new file mode 100644 index 000000000..93eff7855 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolConstants.cs @@ -0,0 +1,61 @@ +// 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.Protocols.Utilities +{ + // ReSharper disable once InconsistentNaming + public static class TJSONProtocolConstants + { + //TODO Check for performance for reusing ImmutableArray from System.Collections.Immutable (https://blogs.msdn.microsoft.com/dotnet/2013/06/24/please-welcome-immutablearrayt/) + // can be possible to get better performance and also better GC + + public static readonly byte[] Comma = {(byte) ','}; + public static readonly byte[] Colon = {(byte) ':'}; + public static readonly byte[] LeftBrace = {(byte) '{'}; + public static readonly byte[] RightBrace = {(byte) '}'}; + public static readonly byte[] LeftBracket = {(byte) '['}; + public static readonly byte[] RightBracket = {(byte) ']'}; + public static readonly byte[] Quote = {(byte) '"'}; + public static readonly byte[] Backslash = {(byte) '\\'}; + + public static readonly byte[] JsonCharTable = + { + 0, 0, 0, 0, 0, 0, 0, 0, (byte) 'b', (byte) 't', (byte) 'n', 0, (byte) 'f', (byte) 'r', 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 1, 1, (byte) '"', 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 + }; + + public static readonly char[] EscapeChars = "\"\\/bfnrt".ToCharArray(); + public static readonly byte[] EscapeCharValues = {(byte) '"', (byte) '\\', (byte) '/', (byte) '\b', (byte) '\f', (byte) '\n', (byte) '\r', (byte) '\t'}; + public static readonly byte[] EscSequences = {(byte) '\\', (byte) 'u', (byte) '0', (byte) '0'}; + + public static class TypeNames + { + public static readonly byte[] NameBool = { (byte)'t', (byte)'f' }; + public static readonly byte[] NameByte = { (byte)'i', (byte)'8' }; + public static readonly byte[] NameI16 = { (byte)'i', (byte)'1', (byte)'6' }; + public static readonly byte[] NameI32 = { (byte)'i', (byte)'3', (byte)'2' }; + public static readonly byte[] NameI64 = { (byte)'i', (byte)'6', (byte)'4' }; + public static readonly byte[] NameDouble = { (byte)'d', (byte)'b', (byte)'l' }; + public static readonly byte[] NameStruct = { (byte)'r', (byte)'e', (byte)'c' }; + public static readonly byte[] NameString = { (byte)'s', (byte)'t', (byte)'r' }; + public static readonly byte[] NameMap = { (byte)'m', (byte)'a', (byte)'p' }; + public static readonly byte[] NameList = { (byte)'l', (byte)'s', (byte)'t' }; + public static readonly byte[] NameSet = { (byte)'s', (byte)'e', (byte)'t' }; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolHelper.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolHelper.cs new file mode 100644 index 000000000..adc26a9af --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TJsonProtocolHelper.cs @@ -0,0 +1,176 @@ +// 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 Thrift.Protocols.Entities; + +namespace Thrift.Protocols.Utilities +{ + // ReSharper disable once InconsistentNaming + public static class TJSONProtocolHelper + { + public static byte[] GetTypeNameForTypeId(TType typeId) + { + switch (typeId) + { + case TType.Bool: + return TJSONProtocolConstants.TypeNames.NameBool; + case TType.Byte: + return TJSONProtocolConstants.TypeNames.NameByte; + case TType.I16: + return TJSONProtocolConstants.TypeNames.NameI16; + case TType.I32: + return TJSONProtocolConstants.TypeNames.NameI32; + case TType.I64: + return TJSONProtocolConstants.TypeNames.NameI64; + case TType.Double: + return TJSONProtocolConstants.TypeNames.NameDouble; + case TType.String: + return TJSONProtocolConstants.TypeNames.NameString; + case TType.Struct: + return TJSONProtocolConstants.TypeNames.NameStruct; + case TType.Map: + return TJSONProtocolConstants.TypeNames.NameMap; + case TType.Set: + return TJSONProtocolConstants.TypeNames.NameSet; + case TType.List: + return TJSONProtocolConstants.TypeNames.NameList; + default: + throw new TProtocolException(TProtocolException.NOT_IMPLEMENTED, "Unrecognized exType"); + } + } + + public static TType GetTypeIdForTypeName(byte[] name) + { + var result = TType.Stop; + if (name.Length > 1) + { + switch (name[0]) + { + case (byte) 'd': + result = TType.Double; + break; + case (byte) 'i': + switch (name[1]) + { + case (byte) '8': + result = TType.Byte; + break; + case (byte) '1': + result = TType.I16; + break; + case (byte) '3': + result = TType.I32; + break; + case (byte) '6': + result = TType.I64; + break; + } + break; + case (byte) 'l': + result = TType.List; + break; + case (byte) 'm': + result = TType.Map; + break; + case (byte) 'r': + result = TType.Struct; + break; + case (byte) 's': + if (name[1] == (byte) 't') + { + result = TType.String; + } + else if (name[1] == (byte) 'e') + { + result = TType.Set; + } + break; + case (byte) 't': + result = TType.Bool; + break; + } + } + if (result == TType.Stop) + { + throw new TProtocolException(TProtocolException.NOT_IMPLEMENTED, "Unrecognized exType"); + } + return result; + } + + /// <summary> + /// Return true if the given byte could be a valid part of a JSON number. + /// </summary> + public static bool IsJsonNumeric(byte b) + { + switch (b) + { + case (byte)'+': + case (byte)'-': + case (byte)'.': + case (byte)'0': + case (byte)'1': + case (byte)'2': + case (byte)'3': + case (byte)'4': + case (byte)'5': + case (byte)'6': + case (byte)'7': + case (byte)'8': + case (byte)'9': + case (byte)'E': + case (byte)'e': + return true; + default: + return false; + } + } + + /// <summary> + /// Convert a byte containing a hex char ('0'-'9' or 'a'-'f') into its + /// corresponding hex value + /// </summary> + public static byte ToHexVal(byte ch) + { + if (ch >= '0' && ch <= '9') + { + return (byte)((char)ch - '0'); + } + + if (ch >= 'a' && ch <= 'f') + { + ch += 10; + return (byte)((char)ch - 'a'); + } + + throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected hex character"); + } + + /// <summary> + /// Convert a byte containing a hex value to its corresponding hex character + /// </summary> + public static byte ToHexChar(byte val) + { + val &= 0x0F; + if (val < 10) + { + return (byte)((char)val + '0'); + } + val -= 10; + return (byte)((char)val + 'a'); + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TProtocolUtil.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TProtocolUtil.cs new file mode 100644 index 000000000..50b038566 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Protocols/Utilities/TProtocolUtil.cs @@ -0,0 +1,110 @@ +// 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; +using Thrift.Protocols.Entities; + +namespace Thrift.Protocols.Utilities +{ + // ReSharper disable once InconsistentNaming + public static class TProtocolUtil + { + public static async Task SkipAsync(TProtocol protocol, TType type, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + protocol.IncrementRecursionDepth(); + try + { + switch (type) + { + case TType.Bool: + await protocol.ReadBoolAsync(cancellationToken); + break; + case TType.Byte: + await protocol.ReadByteAsync(cancellationToken); + break; + case TType.I16: + await protocol.ReadI16Async(cancellationToken); + break; + case TType.I32: + await protocol.ReadI32Async(cancellationToken); + break; + case TType.I64: + await protocol.ReadI64Async(cancellationToken); + break; + case TType.Double: + await protocol.ReadDoubleAsync(cancellationToken); + break; + case TType.String: + // Don't try to decode the string, just skip it. + await protocol.ReadBinaryAsync(cancellationToken); + break; + case TType.Struct: + await protocol.ReadStructBeginAsync(cancellationToken); + while (true) + { + var field = await protocol.ReadFieldBeginAsync(cancellationToken); + if (field.Type == TType.Stop) + { + break; + } + await SkipAsync(protocol, field.Type, cancellationToken); + await protocol.ReadFieldEndAsync(cancellationToken); + } + await protocol.ReadStructEndAsync(cancellationToken); + break; + case TType.Map: + var map = await protocol.ReadMapBeginAsync(cancellationToken); + for (var i = 0; i < map.Count; i++) + { + await SkipAsync(protocol, map.KeyType, cancellationToken); + await SkipAsync(protocol, map.ValueType, cancellationToken); + } + await protocol.ReadMapEndAsync(cancellationToken); + break; + case TType.Set: + var set = await protocol.ReadSetBeginAsync(cancellationToken); + for (var i = 0; i < set.Count; i++) + { + await SkipAsync(protocol, set.ElementType, cancellationToken); + } + await protocol.ReadSetEndAsync(cancellationToken); + break; + case TType.List: + var list = await protocol.ReadListBeginAsync(cancellationToken); + for (var i = 0; i < list.Count; i++) + { + await SkipAsync(protocol, list.ElementType, cancellationToken); + } + await protocol.ReadListEndAsync(cancellationToken); + break; + default: + throw new TProtocolException(TProtocolException.INVALID_DATA, "Unknown data type " + type.ToString("d")); + } + } + finally + { + protocol.DecrementRecursionDepth(); + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Server/AsyncBaseServer.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/AsyncBaseServer.cs new file mode 100644 index 000000000..325c39c71 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/AsyncBaseServer.cs @@ -0,0 +1,183 @@ +// 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.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Thrift.Protocols; +using Thrift.Transports; + +namespace Thrift.Server +{ + //TODO: unhandled exceptions, etc. + + // ReSharper disable once InconsistentNaming + public class AsyncBaseServer : TBaseServer + { + private readonly int _clientWaitingDelay; + private volatile Task _serverTask; + + public AsyncBaseServer(ITAsyncProcessor processor, TServerTransport serverTransport, + ITProtocolFactory inputProtocolFactory, ITProtocolFactory outputProtocolFactory, + ILoggerFactory loggerFactory, int clientWaitingDelay = 10) + : this(new SingletonTProcessorFactory(processor), serverTransport, + new TTransportFactory(), new TTransportFactory(), + inputProtocolFactory, outputProtocolFactory, + loggerFactory.CreateLogger(nameof(AsyncBaseServer)), clientWaitingDelay) + { + } + + public AsyncBaseServer(ITProcessorFactory itProcessorFactory, TServerTransport serverTransport, + TTransportFactory inputTransportFactory, TTransportFactory outputTransportFactory, + ITProtocolFactory inputProtocolFactory, ITProtocolFactory outputProtocolFactory, + ILogger logger, int clientWaitingDelay = 10) + : base(itProcessorFactory, serverTransport, inputTransportFactory, outputTransportFactory, + inputProtocolFactory, outputProtocolFactory, logger) + { + _clientWaitingDelay = clientWaitingDelay; + } + + public override async Task ServeAsync(CancellationToken cancellationToken) + { + try + { + // cancelation token + _serverTask = Task.Factory.StartNew(() => StartListening(cancellationToken), TaskCreationOptions.LongRunning); + await _serverTask; + } + catch (Exception ex) + { + Logger.LogError(ex.ToString()); + } + } + + private async Task StartListening(CancellationToken cancellationToken) + { + ServerTransport.Listen(); + + Logger.LogTrace("Started listening at server"); + + if (ServerEventHandler != null) + { + await ServerEventHandler.PreServeAsync(cancellationToken); + } + + while (!cancellationToken.IsCancellationRequested) + { + if (ServerTransport.IsClientPending()) + { + Logger.LogTrace("Waiting for client connection"); + + try + { + var client = await ServerTransport.AcceptAsync(cancellationToken); + await Task.Factory.StartNew(() => Execute(client, cancellationToken), cancellationToken); + } + catch (TTransportException ttx) + { + Logger.LogTrace($"Transport exception: {ttx}"); + + if (ttx.Type != TTransportException.ExceptionType.Interrupted) + { + Logger.LogError(ttx.ToString()); + } + } + } + else + { + try + { + await Task.Delay(TimeSpan.FromMilliseconds(_clientWaitingDelay), cancellationToken); + } + catch(TaskCanceledException) { } + } + } + + ServerTransport.Close(); + + Logger.LogTrace("Completed listening at server"); + } + + public override void Stop() + { + } + + private async Task Execute(TClientTransport client, CancellationToken cancellationToken) + { + Logger.LogTrace("Started client request processing"); + + var processor = ItProcessorFactory.GetAsyncProcessor(client, this); + + TClientTransport inputTransport = null; + TClientTransport outputTransport = null; + TProtocol inputProtocol = null; + TProtocol outputProtocol = null; + object connectionContext = null; + + try + { + inputTransport = InputTransportFactory.GetTransport(client); + outputTransport = OutputTransportFactory.GetTransport(client); + + inputProtocol = InputProtocolFactory.GetProtocol(inputTransport); + outputProtocol = OutputProtocolFactory.GetProtocol(outputTransport); + + if (ServerEventHandler != null) + { + connectionContext = await ServerEventHandler.CreateContextAsync(inputProtocol, outputProtocol, cancellationToken); + } + + while (!cancellationToken.IsCancellationRequested) + { + if (!await inputTransport.PeekAsync(cancellationToken)) + { + break; + } + + if (ServerEventHandler != null) + { + await ServerEventHandler.ProcessContextAsync(connectionContext, inputTransport, cancellationToken); + } + + if (!await processor.ProcessAsync(inputProtocol, outputProtocol, cancellationToken)) + { + break; + } + } + } + catch (TTransportException ttx) + { + Logger.LogTrace($"Transport exception: {ttx}"); + } + catch (Exception x) + { + Logger.LogError($"Error: {x}"); + } + + if (ServerEventHandler != null) + { + await ServerEventHandler.DeleteContextAsync(connectionContext, inputProtocol, outputProtocol, cancellationToken); + } + + inputTransport?.Close(); + outputTransport?.Close(); + + Logger.LogTrace("Completed client request processing"); + } + } +} diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TBaseServer.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TBaseServer.cs new file mode 100644 index 000000000..741dd5c95 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TBaseServer.cs @@ -0,0 +1,79 @@ +// 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.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Thrift.Protocols; +using Thrift.Transports; + +namespace Thrift.Server +{ + // ReSharper disable once InconsistentNaming + public abstract class TBaseServer + { + protected readonly ILogger Logger; + protected ITProtocolFactory InputProtocolFactory; + protected TTransportFactory InputTransportFactory; + protected ITProcessorFactory ItProcessorFactory; + protected ITProtocolFactory OutputProtocolFactory; + protected TTransportFactory OutputTransportFactory; + + protected TServerEventHandler ServerEventHandler; + protected TServerTransport ServerTransport; + + protected TBaseServer(ITProcessorFactory itProcessorFactory, TServerTransport serverTransport, + TTransportFactory inputTransportFactory, TTransportFactory outputTransportFactory, + ITProtocolFactory inputProtocolFactory, ITProtocolFactory outputProtocolFactory, + ILogger logger) + { + ItProcessorFactory = itProcessorFactory ?? throw new ArgumentNullException(nameof(itProcessorFactory)); + ServerTransport = serverTransport; + InputTransportFactory = inputTransportFactory ?? throw new ArgumentNullException(nameof(inputTransportFactory)); + OutputTransportFactory = outputTransportFactory ?? throw new ArgumentNullException(nameof(outputTransportFactory)); + InputProtocolFactory = inputProtocolFactory ?? throw new ArgumentNullException(nameof(inputProtocolFactory)); + OutputProtocolFactory = outputProtocolFactory ?? throw new ArgumentNullException(nameof(outputProtocolFactory)); + Logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + public void SetEventHandler(TServerEventHandler seh) + { + ServerEventHandler = seh; + } + + public TServerEventHandler GetEventHandler() + { + return ServerEventHandler; + } + + public abstract void Stop(); + + public virtual void Start() + { + // do nothing + } + + public virtual async Task ServeAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TServerEventHandler.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TServerEventHandler.cs new file mode 100644 index 000000000..733bb4bef --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Server/TServerEventHandler.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; +using Thrift.Protocols; +using Thrift.Transports; + +namespace Thrift.Server +{ + //TODO: replacement by event? + + /// <summary> + /// Interface implemented by server users to handle events from the server + /// </summary> + // ReSharper disable once InconsistentNaming + public interface TServerEventHandler + { + /// <summary> + /// Called before the server begins */ + /// </summary> + Task PreServeAsync(CancellationToken cancellationToken); + + /// <summary> + /// Called when a new client has connected and is about to being processing */ + /// </summary> + Task<object> CreateContextAsync(TProtocol input, TProtocol output, CancellationToken cancellationToken); + + /// <summary> + /// Called when a client has finished request-handling to delete server context */ + /// </summary> + Task DeleteContextAsync(object serverContext, TProtocol input, TProtocol output, + CancellationToken cancellationToken); + + /// <summary> + /// Called when a client is about to call the processor */ + /// </summary> + Task ProcessContextAsync(object serverContext, TClientTransport transport, CancellationToken cancellationToken); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/SingletonTProcessorFactory.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/SingletonTProcessorFactory.cs new file mode 100644 index 000000000..c35123348 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/SingletonTProcessorFactory.cs @@ -0,0 +1,38 @@ +// 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 Thrift.Server; +using Thrift.Transports; + +namespace Thrift +{ + // ReSharper disable once InconsistentNaming + public class SingletonTProcessorFactory : ITProcessorFactory + { + private readonly ITAsyncProcessor _tAsyncProcessor; + + public SingletonTProcessorFactory(ITAsyncProcessor tAsyncProcessor) + { + _tAsyncProcessor = tAsyncProcessor; + } + + public ITAsyncProcessor GetAsyncProcessor(TClientTransport trans, TBaseServer baseServer = null) + { + return _tAsyncProcessor; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/TApplicationException.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/TApplicationException.cs new file mode 100644 index 000000000..9ec145a85 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/TApplicationException.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.Threading; +using System.Threading.Tasks; +using Thrift.Protocols; +using Thrift.Protocols.Entities; +using Thrift.Protocols.Utilities; + +namespace Thrift +{ + // ReSharper disable once InconsistentNaming + public class TApplicationException : TException + { + public enum ExceptionType + { + Unknown, + UnknownMethod, + InvalidMessageType, + WrongMethodName, + BadSequenceId, + MissingResult, + InternalError, + ProtocolError, + InvalidTransform, + InvalidProtocol, + UnsupportedClientType + } + + private const int MessageTypeFieldId = 1; + private const int ExTypeFieldId = 2; + + protected ExceptionType Type; + + public TApplicationException() + { + } + + public TApplicationException(ExceptionType type) + { + Type = type; + } + + public TApplicationException(ExceptionType type, string message) + : base(message) + { + Type = type; + } + + public static async Task<TApplicationException> ReadAsync(TProtocol inputProtocol, CancellationToken cancellationToken) + { + string message = null; + var type = ExceptionType.Unknown; + + await inputProtocol.ReadStructBeginAsync(cancellationToken); + while (true) + { + var field = await inputProtocol.ReadFieldBeginAsync(cancellationToken); + if (field.Type == TType.Stop) + { + break; + } + + switch (field.ID) + { + case MessageTypeFieldId: + if (field.Type == TType.String) + { + message = await inputProtocol.ReadStringAsync(cancellationToken); + } + else + { + await TProtocolUtil.SkipAsync(inputProtocol, field.Type, cancellationToken); + } + break; + case ExTypeFieldId: + if (field.Type == TType.I32) + { + type = (ExceptionType) await inputProtocol.ReadI32Async(cancellationToken); + } + else + { + await TProtocolUtil.SkipAsync(inputProtocol, field.Type, cancellationToken); + } + break; + default: + await TProtocolUtil.SkipAsync(inputProtocol, field.Type, cancellationToken); + break; + } + + await inputProtocol.ReadFieldEndAsync(cancellationToken); + } + + await inputProtocol.ReadStructEndAsync(cancellationToken); + + return new TApplicationException(type, message); + } + + public async Task WriteAsync(TProtocol outputProtocol, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + const string messageTypeFieldName = "message"; + const string exTypeFieldName = "exType"; + const string structApplicationExceptionName = "TApplicationException"; + + var struc = new TStruct(structApplicationExceptionName); + var field = new TField(); + + await outputProtocol.WriteStructBeginAsync(struc, cancellationToken); + + if (!string.IsNullOrEmpty(Message)) + { + field.Name = messageTypeFieldName; + field.Type = TType.String; + field.ID = MessageTypeFieldId; + await outputProtocol.WriteFieldBeginAsync(field, cancellationToken); + await outputProtocol.WriteStringAsync(Message, cancellationToken); + await outputProtocol.WriteFieldEndAsync(cancellationToken); + } + + field.Name = exTypeFieldName; + field.Type = TType.I32; + field.ID = ExTypeFieldId; + + await outputProtocol.WriteFieldBeginAsync(field, cancellationToken); + await outputProtocol.WriteI32Async((int) Type, cancellationToken); + await outputProtocol.WriteFieldEndAsync(cancellationToken); + await outputProtocol.WriteFieldStopAsync(cancellationToken); + await outputProtocol.WriteStructEndAsync(cancellationToken); + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/TBaseClient.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/TBaseClient.cs new file mode 100644 index 000000000..e01925153 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/TBaseClient.cs @@ -0,0 +1,91 @@ +// 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.Threading; +using System.Threading.Tasks; +using Thrift.Protocols; + +namespace Thrift +{ + // ReSharper disable once InconsistentNaming + /// <summary> + /// TBaseClient. + /// Base client for generated clients. + /// Do not change this class without checking generated code (namings, etc.) + /// </summary> + public abstract class TBaseClient + { + private readonly TProtocol _inputProtocol; + private readonly TProtocol _outputProtocol; + private bool _isDisposed; + private int _seqId; + public readonly Guid ClientId = Guid.NewGuid(); + + protected TBaseClient(TProtocol inputProtocol, TProtocol outputProtocol) + { + _inputProtocol = inputProtocol ?? throw new ArgumentNullException(nameof(inputProtocol)); + _outputProtocol = outputProtocol ?? throw new ArgumentNullException(nameof(outputProtocol)); + } + + public TProtocol InputProtocol => _inputProtocol; + + public TProtocol OutputProtocol => _outputProtocol; + + public int SeqId + { + get { return ++_seqId; } + } + + public virtual async Task OpenTransportAsync() + { + await OpenTransportAsync(CancellationToken.None); + } + + public virtual async Task OpenTransportAsync(CancellationToken cancellationToken) + { + if (!_inputProtocol.Transport.IsOpen) + { + await _inputProtocol.Transport.OpenAsync(cancellationToken); + } + + if (!_inputProtocol.Transport.IsOpen) + { + await _outputProtocol.Transport.OpenAsync(cancellationToken); + } + } + + public void Dispose() + { + Dispose(true); + } + + protected virtual void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + _inputProtocol?.Dispose(); + _outputProtocol?.Dispose(); + } + } + + _isDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/TException.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/TException.cs new file mode 100644 index 000000000..6aa588d7f --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/TException.cs @@ -0,0 +1,34 @@ +// 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 +{ + // ReSharper disable once InconsistentNaming + public class TException : Exception + { + public TException() + { + } + + public TException(string message) + : base(message) + { + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/TMultiplexedProcessor.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/TMultiplexedProcessor.cs new file mode 100644 index 000000000..ad0e749e0 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/TMultiplexedProcessor.cs @@ -0,0 +1,143 @@ +// 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.Threading; +using System.Threading.Tasks; +using Thrift.Protocols; +using Thrift.Protocols.Entities; + +namespace Thrift +{ + // ReSharper disable once InconsistentNaming + public class TMultiplexedProcessor : ITAsyncProcessor + { + //TODO: Localization + + private readonly Dictionary<string, ITAsyncProcessor> _serviceProcessorMap = + new Dictionary<string, ITAsyncProcessor>(); + + public async Task<bool> ProcessAsync(TProtocol iprot, TProtocol oprot) + { + return await ProcessAsync(iprot, oprot, CancellationToken.None); + } + + public async Task<bool> ProcessAsync(TProtocol iprot, TProtocol oprot, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<bool>(cancellationToken); + } + + try + { + var message = await iprot.ReadMessageBeginAsync(cancellationToken); + + if ((message.Type != TMessageType.Call) && (message.Type != TMessageType.Oneway)) + { + await FailAsync(oprot, message, TApplicationException.ExceptionType.InvalidMessageType, + "Message exType CALL or ONEWAY expected", cancellationToken); + return false; + } + + // Extract the service name + var index = message.Name.IndexOf(TMultiplexedProtocol.Separator, StringComparison.Ordinal); + if (index < 0) + { + await FailAsync(oprot, message, TApplicationException.ExceptionType.InvalidProtocol, + $"Service name not found in message name: {message.Name}. Did you forget to use a TMultiplexProtocol in your client?", + cancellationToken); + return false; + } + + // Create a new TMessage, something that can be consumed by any TProtocol + var serviceName = message.Name.Substring(0, index); + ITAsyncProcessor actualProcessor; + if (!_serviceProcessorMap.TryGetValue(serviceName, out actualProcessor)) + { + await FailAsync(oprot, message, TApplicationException.ExceptionType.InternalError, + $"Service name not found: {serviceName}. Did you forget to call RegisterProcessor()?", + cancellationToken); + return false; + } + + // Create a new TMessage, removing the service name + var newMessage = new TMessage( + message.Name.Substring(serviceName.Length + TMultiplexedProtocol.Separator.Length), + message.Type, + message.SeqID); + + // Dispatch processing to the stored processor + return + await + actualProcessor.ProcessAsync(new StoredMessageProtocol(iprot, newMessage), oprot, + cancellationToken); + } + catch (IOException) + { + return false; // similar to all other processors + } + } + + public void RegisterProcessor(string serviceName, ITAsyncProcessor processor) + { + if (_serviceProcessorMap.ContainsKey(serviceName)) + { + throw new InvalidOperationException( + $"Processor map already contains processor with name: '{serviceName}'"); + } + + _serviceProcessorMap.Add(serviceName, processor); + } + + private async Task FailAsync(TProtocol oprot, TMessage message, TApplicationException.ExceptionType extype, + string etxt, CancellationToken cancellationToken) + { + var appex = new TApplicationException(extype, etxt); + + var newMessage = new TMessage(message.Name, TMessageType.Exception, message.SeqID); + + await oprot.WriteMessageBeginAsync(newMessage, cancellationToken); + await appex.WriteAsync(oprot, cancellationToken); + await oprot.WriteMessageEndAsync(cancellationToken); + await oprot.Transport.FlushAsync(cancellationToken); + } + + private class StoredMessageProtocol : TProtocolDecorator + { + readonly TMessage _msgBegin; + + public StoredMessageProtocol(TProtocol protocol, TMessage messageBegin) + : base(protocol) + { + _msgBegin = messageBegin; + } + + public override async Task<TMessage> ReadMessageBeginAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TMessage>(cancellationToken); + } + + return _msgBegin; + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Thrift.csproj b/src/jaegertracing/thrift/lib/netcore/Thrift/Thrift.csproj new file mode 100644 index 000000000..6a2ccc30d --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Thrift.csproj @@ -0,0 +1,33 @@ +<Project Sdk="Microsoft.NET.Sdk"> + + <PropertyGroup> + <TargetFramework>netstandard2.0</TargetFramework> + <AssemblyName>Thrift</AssemblyName> + <PackageId>Thrift</PackageId> + <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects> + <GenerateBindingRedirectsOutputType>true</GenerateBindingRedirectsOutputType> + <GenerateAssemblyTitleAttribute>false</GenerateAssemblyTitleAttribute> + <GenerateAssemblyDescriptionAttribute>false</GenerateAssemblyDescriptionAttribute> + <GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute> + <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute> + <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute> + <GenerateAssemblyCopyrightAttribute>false</GenerateAssemblyCopyrightAttribute> + <GenerateAssemblyVersionAttribute>false</GenerateAssemblyVersionAttribute> + <GenerateAssemblyFileVersionAttribute>false</GenerateAssemblyFileVersionAttribute> + <SignAssembly>true</SignAssembly> + <AssemblyOriginatorKeyFile>thrift.snk</AssemblyOriginatorKeyFile> + </PropertyGroup> + + <ItemGroup> + <PackageReference Include="Microsoft.AspNetCore" Version="[2.0,)" /> + <PackageReference Include="Microsoft.Extensions.Logging" Version="[2.0,)" /> + <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="[2.0,)" /> + <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="[2.0,)" /> + <PackageReference Include="System.IO.Pipes" Version="[4.3,)" /> + <PackageReference Include="System.Net.Http.WinHttpHandler" Version="4.4.0" /> + <PackageReference Include="System.Net.NameResolution" Version="[4.3,)" /> + <PackageReference Include="System.Net.Requests" Version="[4.3,)" /> + <PackageReference Include="System.Net.Security" Version="[4.3,)" /> + </ItemGroup> + +</Project> diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TBufferedClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TBufferedClientTransport.cs new file mode 100644 index 000000000..761f1ac78 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TBufferedClientTransport.cs @@ -0,0 +1,206 @@ +// 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.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class TBufferedClientTransport : TClientTransport + { + private readonly int _bufSize; + private readonly MemoryStream _inputBuffer = new MemoryStream(0); + private readonly MemoryStream _outputBuffer = new MemoryStream(0); + private readonly TClientTransport _transport; + private bool _isDisposed; + + //TODO: should support only specified input transport? + public TBufferedClientTransport(TClientTransport transport, int bufSize = 1024) + { + if (bufSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(bufSize), "Buffer size must be a positive number."); + } + + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + _bufSize = bufSize; + } + + public TClientTransport UnderlyingTransport + { + get + { + CheckNotDisposed(); + + return _transport; + } + } + + public override bool IsOpen => !_isDisposed && _transport.IsOpen; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + await _transport.OpenAsync(cancellationToken); + } + + public override void Close() + { + CheckNotDisposed(); + + _transport.Close(); + } + + public override async Task<int> ReadAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + //TODO: investigate how it should work correctly + CheckNotDisposed(); + + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + if (_inputBuffer.Capacity < _bufSize) + { + _inputBuffer.Capacity = _bufSize; + } + + var got = await _inputBuffer.ReadAsync(buffer, offset, length, cancellationToken); + if (got > 0) + { + return got; + } + + _inputBuffer.Seek(0, SeekOrigin.Begin); + _inputBuffer.SetLength(_inputBuffer.Capacity); + + ArraySegment<byte> bufSegment; + _inputBuffer.TryGetBuffer(out bufSegment); + + // investigate + var filled = await _transport.ReadAsync(bufSegment.Array, 0, (int) _inputBuffer.Length, cancellationToken); + _inputBuffer.SetLength(filled); + + if (filled == 0) + { + return 0; + } + + return await 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); + } + + // Relative offset from "off" argument + var writtenCount = 0; + if (_outputBuffer.Length > 0) + { + var capa = (int) (_outputBuffer.Capacity - _outputBuffer.Length); + var writeSize = capa <= length ? capa : length; + await _outputBuffer.WriteAsync(buffer, offset, writeSize, cancellationToken); + + writtenCount += writeSize; + if (writeSize == capa) + { + //ArraySegment<byte> bufSegment; + //_outputBuffer.TryGetBuffer(out bufSegment); + var data = _outputBuffer.ToArray(); + //await _transport.WriteAsync(bufSegment.Array, cancellationToken); + await _transport.WriteAsync(data, cancellationToken); + _outputBuffer.SetLength(0); + } + } + + while (length - writtenCount >= _bufSize) + { + await _transport.WriteAsync(buffer, offset + writtenCount, _bufSize, cancellationToken); + writtenCount += _bufSize; + } + + var remain = length - writtenCount; + if (remain > 0) + { + if (_outputBuffer.Capacity < _bufSize) + { + _outputBuffer.Capacity = _bufSize; + } + await _outputBuffer.WriteAsync(buffer, offset + writtenCount, remain, cancellationToken); + } + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + if (_outputBuffer.Length > 0) + { + //ArraySegment<byte> bufSegment; + var data = _outputBuffer.ToArray(); // TryGetBuffer(out bufSegment); + + await _transport.WriteAsync(data /*bufSegment.Array*/, cancellationToken); + _outputBuffer.SetLength(0); + } + + await _transport.FlushAsync(cancellationToken); + } + + private void CheckNotDisposed() + { + if (_isDisposed) + { + throw new ObjectDisposedException(nameof(_transport)); + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + _inputBuffer?.Dispose(); + _outputBuffer?.Dispose(); + _transport?.Dispose(); + } + } + _isDisposed = true; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TFramedClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TFramedClientTransport.cs new file mode 100644 index 000000000..d11bb959a --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TFramedClientTransport.cs @@ -0,0 +1,201 @@ +// 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.Transports.Client +{ + //TODO: check for correct implementation + + // ReSharper disable once InconsistentNaming + public class TFramedClientTransport : TClientTransport + { + private const int HeaderSize = 4; + private readonly byte[] _headerBuf = new byte[HeaderSize]; + private readonly MemoryStream _readBuffer = new MemoryStream(1024); + private readonly TClientTransport _transport; + private readonly MemoryStream _writeBuffer = new MemoryStream(1024); + + private bool _isDisposed; + + public TFramedClientTransport(TClientTransport transport) + { + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + + InitWriteBuffer(); + } + + public override bool IsOpen => !_isDisposed && _transport.IsOpen; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + await _transport.OpenAsync(cancellationToken); + } + + public override void Close() + { + CheckNotDisposed(); + + _transport.Close(); + } + + public override async Task<int> ReadAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + CheckNotDisposed(); + + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + var got = await _readBuffer.ReadAsync(buffer, offset, length, cancellationToken); + if (got > 0) + { + return got; + } + + // Read another frame of data + await ReadFrameAsync(cancellationToken); + + return await _readBuffer.ReadAsync(buffer, offset, length, cancellationToken); + } + + private async Task ReadFrameAsync(CancellationToken cancellationToken) + { + await _transport.ReadAllAsync(_headerBuf, 0, HeaderSize, cancellationToken); + + var size = DecodeFrameSize(_headerBuf); + + _readBuffer.SetLength(size); + _readBuffer.Seek(0, SeekOrigin.Begin); + + ArraySegment<byte> bufSegment; + _readBuffer.TryGetBuffer(out bufSegment); + + var buff = bufSegment.Array; + + await _transport.ReadAllAsync(buff, 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 + length > int.MaxValue) + { + 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); + //var buf = bufSegment.Array; + var buf = _writeBuffer.ToArray(); + + //var len = (int)_writeBuffer.Length; + var dataLen = (int) _writeBuffer.Length - HeaderSize; + if (dataLen < 0) + { + throw new InvalidOperationException(); // logic error actually + } + + // Inject message header into the reserved buffer space + EncodeFrameSize(dataLen, buf); + + // Send the entire message at once + await _transport.WriteAsync(buf, cancellationToken); + + InitWriteBuffer(); + + await _transport.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("TFramedClientTransport"); + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + _readBuffer?.Dispose(); + _writeBuffer?.Dispose(); + _transport?.Dispose(); + } + } + _isDisposed = true; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/THttpClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/THttpClientTransport.cs new file mode 100644 index 000000000..8bce9e4a0 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/THttpClientTransport.cs @@ -0,0 +1,227 @@ +// 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.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class THttpClientTransport : TClientTransport + { + private readonly X509Certificate[] _certificates; + private readonly Uri _uri; + + // Timeouts in milliseconds + private int _connectTimeout = 30000; + private HttpClient _httpClient; + private Stream _inputStream; + + private bool _isDisposed; + private MemoryStream _outputStream = new MemoryStream(); + + public THttpClientTransport(Uri u, IDictionary<string, string> customHeaders) + : this(u, Enumerable.Empty<X509Certificate>(), customHeaders) + { + } + + public THttpClientTransport(Uri u, IEnumerable<X509Certificate> certificates, + IDictionary<string, string> customHeaders) + { + _uri = u; + _certificates = (certificates ?? Enumerable.Empty<X509Certificate>()).ToArray(); + CustomHeaders = customHeaders; + + // 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(); + } + + public IDictionary<string, string> CustomHeaders { get; } + + public int ConnectTimeout + { + set { _connectTimeout = value; } + get { return _connectTimeout; } + } + + 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; + } + + if (_httpClient != null) + { + _httpClient.Dispose(); + _httpClient = null; + } + } + + public override async Task<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() + { + var handler = new HttpClientHandler(); + handler.ClientCertificates.AddRange(_certificates); + + 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.Add(new ProductInfoHeaderValue("THttpClientTransport", "0.13.0")); + + if (CustomHeaders != null) + { + foreach (var item in CustomHeaders) + { + httpClient.DefaultRequestHeaders.Add(item.Key, item.Value); + } + } + + return httpClient; + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + try + { + try + { + if (_outputStream.CanSeek) + { + _outputStream.Seek(0, SeekOrigin.Begin); + } + + using (var outStream = new StreamContent(_outputStream)) + { + var msg = await _httpClient.PostAsync(_uri, outStream, cancellationToken); + + msg.EnsureSuccessStatusCode(); + + if (_inputStream != null) + { + _inputStream.Dispose(); + _inputStream = null; + } + + _inputStream = await msg.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); + } + } + 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/netcore/Thrift/Transports/Client/TMemoryBufferClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TMemoryBufferClientTransport.cs new file mode 100644 index 000000000..46a55a64a --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TMemoryBufferClientTransport.cs @@ -0,0 +1,97 @@ +// 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.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class TMemoryBufferClientTransport : TClientTransport + { + private readonly MemoryStream _byteStream; + private bool _isDisposed; + + public TMemoryBufferClientTransport() + { + _byteStream = new MemoryStream(); + } + + public TMemoryBufferClientTransport(byte[] buf) + { + _byteStream = new MemoryStream(buf); + } + + 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 override async Task<int> ReadAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + return await _byteStream.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, CancellationToken cancellationToken) + { + await _byteStream.WriteAsync(buffer, 0, buffer.Length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + await _byteStream.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public byte[] GetBuffer() + { + return _byteStream.ToArray(); + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + _byteStream?.Dispose(); + } + } + _isDisposed = true; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TNamedPipeClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TNamedPipeClientTransport.cs new file mode 100644 index 000000000..f5e4baf4a --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TNamedPipeClientTransport.cs @@ -0,0 +1,95 @@ +// 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.Pipes; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class TNamedPipeClientTransport : TClientTransport + { + private NamedPipeClientStream _client; + + public TNamedPipeClientTransport(string pipe) : this(".", pipe) + { + } + + public TNamedPipeClientTransport(string server, string pipe) + { + var serverName = string.IsNullOrWhiteSpace(server) ? server : "."; + + _client = new NamedPipeClientStream(serverName, pipe, PipeDirection.InOut, PipeOptions.None); + } + + public override bool IsOpen => _client != null && _client.IsConnected; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.AlreadyOpen); + } + + await _client.ConnectAsync(cancellationToken); + } + + public override void Close() + { + if (_client != null) + { + _client.Dispose(); + _client = null; + } + } + + public override async Task<int> ReadAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + if (_client == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + return await _client.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (_client == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + await _client.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + protected override void Dispose(bool disposing) + { + _client.Dispose(); + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TSocketClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TSocketClientTransport.cs new file mode 100644 index 000000000..e769d1421 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TSocketClientTransport.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; + +namespace Thrift.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class TSocketClientTransport : TStreamClientTransport + { + private bool _isDisposed; + + public TSocketClientTransport(TcpClient client) + { + TcpClient = client ?? throw new ArgumentNullException(nameof(client)); + + if (IsOpen) + { + InputStream = client.GetStream(); + OutputStream = client.GetStream(); + } + } + + public TSocketClientTransport(IPAddress host, int port) + : this(host, port, 0) + { + } + + public TSocketClientTransport(IPAddress host, int port, int timeout) + { + Host = host; + Port = port; + + TcpClient = new TcpClient(); + TcpClient.ReceiveTimeout = TcpClient.SendTimeout = timeout; + TcpClient.Client.NoDelay = true; + } + + 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 + { + if (TcpClient == null) + { + return false; + } + + return 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); + + InputStream = TcpClient.GetStream(); + OutputStream = TcpClient.GetStream(); + } + + 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/netcore/Thrift/Transports/Client/TStreamClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TStreamClientTransport.cs new file mode 100644 index 000000000..be2011e83 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TStreamClientTransport.cs @@ -0,0 +1,110 @@ +// 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.Transports.Client +{ + // ReSharper disable once InconsistentNaming + public class TStreamClientTransport : TClientTransport + { + private bool _isDisposed; + + protected TStreamClientTransport() + { + } + + public TStreamClientTransport(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 Task<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; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TTlsSocketClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TTlsSocketClientTransport.cs new file mode 100644 index 000000000..c8be4ede1 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Client/TTlsSocketClientTransport.cs @@ -0,0 +1,237 @@ +// 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.Transports.Client +{ + //TODO: check for correct work + + // ReSharper disable once InconsistentNaming + public class TTlsSocketClientTransport : TStreamClientTransport + { + 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 TTlsSocketClientTransport(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("TTlsSocketClientTransport needs certificate to be used for server", + nameof(certificate)); + } + + if (IsOpen) + { + InputStream = client.GetStream(); + OutputStream = client.GetStream(); + } + } + + public TTlsSocketClientTransport(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 TTlsSocketClientTransport(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 TTlsSocketClientTransport(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 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; + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/THttpServerTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/THttpServerTransport.cs new file mode 100644 index 000000000..032063a37 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/THttpServerTransport.cs @@ -0,0 +1,98 @@ +// 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.Protocols; +using Thrift.Transports.Client; + +namespace Thrift.Transports.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 ITProtocolFactory InputProtocolFactory; + protected ITProtocolFactory OutputProtocolFactory; + + protected ITAsyncProcessor Processor; + + public THttpServerTransport(ITAsyncProcessor processor, RequestDelegate next, ILoggerFactory loggerFactory) + : this(processor, new TBinaryProtocol.Factory(), next, loggerFactory) + { + } + + public THttpServerTransport(ITAsyncProcessor processor, ITProtocolFactory protocolFactory, RequestDelegate next, + ILoggerFactory loggerFactory) + : this(processor, protocolFactory, protocolFactory, next, loggerFactory) + { + } + + public THttpServerTransport(ITAsyncProcessor processor, ITProtocolFactory inputProtocolFactory, + ITProtocolFactory outputProtocolFactory, RequestDelegate next, ILoggerFactory loggerFactory) + { + if (loggerFactory == null) + { + throw new ArgumentNullException(nameof(loggerFactory)); + } + + Processor = processor ?? throw new ArgumentNullException(nameof(processor)); + InputProtocolFactory = inputProtocolFactory ?? throw new ArgumentNullException(nameof(inputProtocolFactory)); + OutputProtocolFactory = outputProtocolFactory ?? throw new ArgumentNullException(nameof(outputProtocolFactory)); + + _next = next; + _logger = loggerFactory.CreateLogger<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 TStreamClientTransport(context.Request.Body, context.Response.Body); + + try + { + var input = InputProtocolFactory.GetProtocol(transport); + var output = OutputProtocolFactory.GetProtocol(transport); + + while (await Processor.ProcessAsync(input, output, cancellationToken)) + { + } + } + catch (TTransportException) + { + // Client died, just move on + } + finally + { + transport.Close(); + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TNamedPipeServerTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TNamedPipeServerTransport.cs new file mode 100644 index 000000000..186786ed2 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TNamedPipeServerTransport.cs @@ -0,0 +1,191 @@ +// 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.Transports.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 + { + //TODO: check for disconection + _stream.Disconnect(); + _stream.Dispose(); + } + finally + { + _stream = null; + _isPending = false; + } + } + } + + public override bool IsClientPending() + { + return _isPending; + } + + private void EnsurePipeInstance() + { + if (_stream == null) + { + var direction = PipeDirection.InOut; + var maxconn = 254; + var mode = PipeTransmissionMode.Byte; + var options = _asyncMode ? PipeOptions.Asynchronous : PipeOptions.None; + var inbuf = 4096; + var outbuf = 4096; + // TODO: security + + try + { + _stream = new NamedPipeServerStream(_pipeAddress, direction, maxconn, mode, options, inbuf, outbuf); + } + 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; + } + } + } + } + + protected override async Task<TClientTransport> 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 : TClientTransport + { + private readonly NamedPipeServerStream _stream; + + public ServerTransport(NamedPipeServerStream stream) + { + _stream = stream; + } + + public override bool IsOpen => _stream != null && _stream.IsConnected; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override void Close() + { + _stream?.Dispose(); + } + + public override async Task<int> ReadAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + if (_stream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + return await _stream.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + if (_stream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + await _stream.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + protected override void Dispose(bool disposing) + { + _stream?.Dispose(); + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerFramedTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerFramedTransport.cs new file mode 100644 index 000000000..0b86e9ebb --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerFramedTransport.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.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Thrift.Transports.Client; + +namespace Thrift.Transports.Server +{ + // ReSharper disable once InconsistentNaming + public class TServerFramedTransport : TServerTransport + { + private readonly int _clientTimeout; + private readonly int _port; + private TcpListener _server; + + public TServerFramedTransport(TcpListener listener) + : this(listener, 0) + { + } + + public TServerFramedTransport(TcpListener listener, int clientTimeout) + { + _server = listener; + _clientTimeout = clientTimeout; + } + + public TServerFramedTransport(int port) + : this(port, 0) + { + } + + public TServerFramedTransport(int port, int clientTimeout) + { + _port = port; + _clientTimeout = 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 Task<TClientTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TClientTransport>(cancellationToken); + } + + if (_server == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "No underlying server socket."); + } + + try + { + TFramedClientTransport tSocketTransport = null; + var tcpClient = await _server.AcceptTcpClientAsync(); + + try + { + tSocketTransport = new TFramedClientTransport(new TSocketClientTransport(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; + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerSocketTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerSocketTransport.cs new file mode 100644 index 000000000..3a9d8a17d --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TServerSocketTransport.cs @@ -0,0 +1,174 @@ +// 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.Transports.Client; + +namespace Thrift.Transports.Server +{ + // ReSharper disable once InconsistentNaming + public class TServerSocketTransport : TServerTransport + { + private readonly int _clientTimeout; + private readonly int _port; + private readonly bool _useBufferedSockets; + private readonly bool _useFramedTransport; + private TcpListener _server; + + public TServerSocketTransport(TcpListener listener) + : this(listener, 0) + { + } + + public TServerSocketTransport(TcpListener listener, int clientTimeout) + { + _server = listener; + _clientTimeout = clientTimeout; + } + + public TServerSocketTransport(int port) + : this(port, 0) + { + } + + public TServerSocketTransport(int port, int clientTimeout) + : this(port, clientTimeout, false) + { + } + + public TServerSocketTransport(int port, int clientTimeout, bool useBufferedSockets): + this(port, clientTimeout, useBufferedSockets, false) + { + } + + public TServerSocketTransport(int port, int clientTimeout, bool useBufferedSockets, bool useFramedTransport) + { + _port = port; + _clientTimeout = clientTimeout; + _useBufferedSockets = useBufferedSockets; + _useFramedTransport = useFramedTransport; + 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 Task<TClientTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TClientTransport>(cancellationToken); + } + + if (_server == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "No underlying server socket."); + } + + try + { + TClientTransport tSocketTransport = null; + var tcpClient = await _server.AcceptTcpClientAsync(); + + try + { + tSocketTransport = new TSocketClientTransport(tcpClient) + { + Timeout = _clientTimeout + }; + + if (_useBufferedSockets) + { + tSocketTransport = new TBufferedClientTransport(tSocketTransport); + } + + if (_useFramedTransport) + { + tSocketTransport = new TFramedClientTransport(tSocketTransport); + } + + 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; + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TTlsServerSocketTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TTlsServerSocketTransport.cs new file mode 100644 index 000000000..759feeddd --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/Server/TTlsServerSocketTransport.cs @@ -0,0 +1,177 @@ +// 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.Transports.Client; + +namespace Thrift.Transports.Server +{ + // ReSharper disable once InconsistentNaming + public class TTlsServerSocketTransport : TServerTransport + { + private readonly RemoteCertificateValidationCallback _clientCertValidator; + private readonly int _clientTimeout = 0; + private readonly LocalCertificateSelectionCallback _localCertificateSelectionCallback; + private readonly int _port; + private readonly X509Certificate2 _serverCertificate; + private readonly SslProtocols _sslProtocols; + private readonly bool _useBufferedSockets; + private readonly bool _useFramedTransport; + private TcpListener _server; + + public TTlsServerSocketTransport(int port, X509Certificate2 certificate) + : this(port, false, certificate) + { + } + + public TTlsServerSocketTransport( + int port, + bool useBufferedSockets, + X509Certificate2 certificate, + RemoteCertificateValidationCallback clientCertValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + : this(port, useBufferedSockets, false, certificate, + clientCertValidator, localCertificateSelectionCallback, sslProtocols) + { + } + + public TTlsServerSocketTransport( + int port, + bool useBufferedSockets, + bool useFramedTransport, + 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"); + } + + _port = port; + _serverCertificate = certificate; + _useBufferedSockets = useBufferedSockets; + _useFramedTransport = useFramedTransport; + _clientCertValidator = clientCertValidator; + _localCertificateSelectionCallback = localCertificateSelectionCallback; + _sslProtocols = sslProtocols; + + 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 Task<TClientTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TClientTransport>(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 TTlsSocketClientTransport(client, _serverCertificate, true, _clientCertValidator, + _localCertificateSelectionCallback, _sslProtocols); + + await tTlsSocket.SetupTlsAsync(); + + TClientTransport trans = tTlsSocket; + + if (_useBufferedSockets) + { + trans = new TBufferedClientTransport(trans); + } + + if (_useFramedTransport) + { + trans = new TFramedClientTransport(trans); + } + + return trans; + } + 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; + } + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TClientTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TClientTransport.cs new file mode 100644 index 000000000..0dd96cb36 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TClientTransport.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.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transports +{ + //TODO: think about client info + // ReSharper disable once InconsistentNaming + public abstract class TClientTransport : 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 Task<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 (offset < 0) + { + throw new ArgumentOutOfRangeException(nameof(offset), "Buffer offset is smaller than zero."); + } + + if (length < 0) + { + throw new ArgumentOutOfRangeException(nameof(length), "Buffer length is smaller than zero."); + } + + if (offset + length > buffer.Length) + { + throw new ArgumentOutOfRangeException(nameof(buffer), "Not enough data."); + } + } + + public virtual async Task<int> ReadAsync(byte[] buffer, int offset, int length) + { + return await ReadAsync(buffer, offset, length, CancellationToken.None); + } + + public abstract Task<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken); + + public virtual async Task<int> ReadAllAsync(byte[] buffer, int offset, int length) + { + return await ReadAllAsync(buffer, offset, length, CancellationToken.None); + } + + public virtual async Task<int> ReadAllAsync(byte[] buffer, int offset, int length, + CancellationToken cancellationToken) + { + ValidateBufferArgs(buffer, offset, length); + + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<int>(cancellationToken); + } + + var retrieved = 0; + + //If we previously peeked a byte, we need to use that first. + if (_hasPeekByte) + { + buffer[offset + retrieved++] = _peekBuffer[0]; + _hasPeekByte = false; + } + + while (retrieved < length) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<int>(cancellationToken); + } + + var returnedCount = await ReadAsync(buffer, offset + retrieved, length - retrieved, cancellationToken); + if (returnedCount <= 0) + { + throw new TTransportException(TTransportException.ExceptionType.EndOfFile, + "Cannot read, Remote side has closed"); + } + retrieved += returnedCount; + } + return retrieved; + } + + 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); + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TServerTransport.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TServerTransport.cs new file mode 100644 index 000000000..0d45a55f9 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/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.Transports +{ + // ReSharper disable once InconsistentNaming + public abstract class TServerTransport + { + public abstract void Listen(); + public abstract void Close(); + public abstract bool IsClientPending(); + + protected virtual async Task<TClientTransport> AcceptImplementationAsync() + { + return await AcceptImplementationAsync(CancellationToken.None); + } + + protected abstract Task<TClientTransport> AcceptImplementationAsync(CancellationToken cancellationToken); + + public async Task<TClientTransport> AcceptAsync() + { + return await AcceptAsync(CancellationToken.None); + } + + public async Task<TClientTransport> 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/netcore/Thrift/Transports/TTransportException.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TTransportException.cs new file mode 100644 index 000000000..b7c42e33a --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TTransportException.cs @@ -0,0 +1,58 @@ +// 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.Transports +{ + // ReSharper disable once InconsistentNaming + public class TTransportException : TException + { + public enum ExceptionType + { + Unknown, + NotOpen, + AlreadyOpen, + TimedOut, + EndOfFile, + Interrupted + } + + protected ExceptionType ExType; + + public TTransportException() + { + } + + public TTransportException(ExceptionType exType) + : this() + { + ExType = exType; + } + + public TTransportException(ExceptionType exType, string message) + : base(message) + { + ExType = exType; + } + + public TTransportException(string message) + : base(message) + { + } + + public ExceptionType Type => ExType; + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TTransportFactory.cs b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/TTransportFactory.cs new file mode 100644 index 000000000..26c3cc471 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/Transports/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.Transports +{ + /// <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 TClientTransport GetTransport(TClientTransport trans) + { + return trans; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netcore/Thrift/thrift.snk b/src/jaegertracing/thrift/lib/netcore/Thrift/thrift.snk Binary files differnew file mode 100644 index 000000000..97bc5812b --- /dev/null +++ b/src/jaegertracing/thrift/lib/netcore/Thrift/thrift.snk |