922 lines
32 KiB
C#
922 lines
32 KiB
C#
// 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;
|
|
}
|
|
}
|
|
} |