Skip to content

Generate Byte Buffer native code for thrift/mux services #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nayato wants to merge 9 commits into
base: csharper
Choose a base branch
from
Open

Generate Byte Buffer native code for thrift/mux services #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
cf4857a
wip buffers
nayato Aug 6, 2018
40c7e2b
1
nayato Aug 6, 2018
54a85c7
abc
nayato Aug 6, 2018
1ebdcde
abc
nayato Aug 6, 2018
c1fc133
abc
nayato Aug 6, 2018
f941663
abc
nayato Aug 6, 2018
fae1b7e
abc
nayato Aug 6, 2018
b1d482b
abc
nayato Aug 6, 2018
f769b8b
abc
nayato Aug 6, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 43 additions & 25 deletions scrooge-generator/src/main/resources/csharpgen/service.mustache
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,31 +16,41 @@

public partial class ServiceToClient{{#extends_iface}} : {{name}}.ServiceToClient, {{/extends_iface}}{{^extends_iface}} : {{/extends_iface}}ServiceIface
{
readonly Finagle.Core.Service<ThriftClientRequest, byte[]> service;
readonly TProtocolFactory protocolFactory;
{{^extends_iface}}
protected readonly Finagle.Core.Service<ThriftClientRequest, IByteBuffer> service;
protected readonly TProtocolFactory protocolFactory;
protected readonly IByteBufferAllocator allocator;
{{/extends_iface}}
{{^is_oneway}}
readonly Finagle.Core.Services.ResponseClassifier responseClassifier;

public ServiceToClient(Finagle.Core.Service<ThriftClientRequest, byte[]> service, TProtocolFactory protocolFactory, Finagle.Core.Services.ResponseClassifier responseClassifier){{#extends_iface}} : base(service, protocolFactory, responseClassifier){{/extends_iface}}
public ServiceToClient(Finagle.Core.Service<ThriftClientRequest, IByteBuffer> service, TProtocolFactory protocolFactory, Finagle.Core.Services.ResponseClassifier responseClassifier, IByteBufferAllocator allocator)
{{#extends_iface}} : base(service, protocolFactory, responseClassifier, allocator){{/extends_iface}}
{
{{^extends_iface}}
this.service = service;
this.protocolFactory = protocolFactory;
this.responseClassifier = responseClassifier;
this.allocator = allocator;
{{/extends_iface}}
}
{{/is_oneway}}

public ServiceToClient(Finagle.Core.Service<ThriftClientRequest, byte[]> service, TProtocolFactory protocolFactory) {{#extends_iface}}: base(service, protocolFactory, Finagle.Core.Services.ResponseClassifier.Default){{/extends_iface}}
public ServiceToClient(Finagle.Core.Service<ThriftClientRequest, IByteBuffer> service, TProtocolFactory protocolFactory, IByteBufferAllocator allocator) {{#extends_iface}}: base(service, protocolFactory, Finagle.Core.Services.ResponseClassifier.Default, allocator){{/extends_iface}}
{
{{^extends_iface}}
this.service = service;
this.protocolFactory = protocolFactory;
this.allocator = allocator;
{{^is_oneway}}this.responseClassifier = Finagle.Core.Services.ResponseClassifier.Default;{{/is_oneway}}
{{/extends_iface}}
}

{{^extends_iface}}
protected T DecodeResponse<T>(byte[] resBytes, ThriftStructCodec3<T> codec)
protected T DecodeResponse<T>(IByteBuffer resBytes, ThriftStructCodec3<T> codec)
where T : IThriftStruct
{
var iprot = this.protocolFactory.GetProtocol(new TMemoryInputTransport(resBytes));
var iprot = this.protocolFactory.GetProtocol(new TBufferInputTransport(resBytes));
var msg = iprot.ReadMessageBegin();
try
{
Expand All @@ -51,6 +61,7 @@
finally
{
iprot.ReadMessageEnd();
iprot.Dispose();
}
}
{{/extends_iface}}
Expand All @@ -61,7 +72,7 @@
public async Task{{^is_oneway_or_void}}<{{{return_type.type_name_in_container}}}>{{/is_oneway_or_void}} {{name}}Async({{{argument_list_with_types}}})
{
// TODO: size
TMemoryBuffer __memoryTransport__ = new TMemoryBuffer();
TByteBufferTransport __memoryTransport__ = new TByteBufferTransport(this.allocator);
TProtocol __prot__ = this.protocolFactory.GetProtocol(__memoryTransport__);
__prot__.WriteMessageBegin(new TMessage("{{name}}", TMessageType.Call, 0));
{{name}}_args __args__ = new {{name}}_args();
Expand All @@ -72,7 +83,7 @@
__prot__.WriteMessageEnd();


byte[] __buffer__ = __memoryTransport__.GetBuffer();
IByteBuffer __buffer__ = __memoryTransport__.GetBuffer();
ThriftClientRequest __request__ = new ThriftClientRequest(__buffer__, {{#is_oneway}}true{{/is_oneway}}{{^is_oneway}}false{{/is_oneway}});
{{^is_oneway_or_void}}
var serdeCtx = new DeserializeCtx<{{{return_type.type_name}}}>(__args__, buf => Decode{{name}}Response(buf));
Expand Down Expand Up @@ -104,7 +115,7 @@
}
{{^is_oneway_or_void}}

Finagle.Core.Util.Try<{{{return_type.type_name}}}> Decode{{name}}Response(byte[] response)
Finagle.Core.Util.Try<{{{return_type.type_name}}}> Decode{{name}}Response(IByteBuffer response)
{
try
{
Expand All @@ -129,33 +140,36 @@
{{/functions}}
}

public partial class Service {{#extends_iface}}: {{name}}.Service{{/extends_iface}}{{^extends_iface}}: Finagle.Core.Service<byte[], byte[]>{{/extends_iface}}
public partial class Service {{#extends_iface}}: {{name}}.Service{{/extends_iface}}{{^extends_iface}}: Finagle.Core.Service<IByteBuffer, IByteBuffer>{{/extends_iface}}
{
readonly ServiceIface iface;
readonly TProtocolFactory protocolFactory;

{{^extends_iface}}
protected IDictionary<string, Func<TProtocol, int, Task<byte[]>>> functionMap = new Dictionary<string, Func<TProtocol, int, Task<byte[]>>>();
protected readonly IByteBufferAllocator allocator;
protected IDictionary<string, Func<TProtocol, int, Task<IByteBuffer>>> functionMap = new Dictionary<string, Func<TProtocol, int, Task<IByteBuffer>>>();
{{/extends_iface}}

byte[] CreateException(string name, int seqid, TApplicationException.ExceptionType code, string message)
IByteBuffer CreateException(string name, int seqid, TApplicationException.ExceptionType code, string message)
{
TApplicationException x = new TApplicationException(code, message);
TMemoryBuffer memoryBuffer = new TMemoryBuffer();
TByteBufferTransport memoryBuffer = new TByteBufferTransport(this.allocator);
TProtocol oprot = this.protocolFactory.GetProtocol(memoryBuffer);

oprot.WriteMessageBegin(new TMessage(name, TMessageType.Exception, seqid));
x.Write(oprot);
oprot.WriteMessageEnd();
oprot.Transport.Flush();
byte[] buffer = memoryBuffer.GetBuffer();
IByteBuffer buffer = memoryBuffer.GetBuffer();
return buffer;
}

public Service(ServiceIface iface, TProtocolFactory protocolFactory){{#extends_iface}} : base(iface, protocolFactory){{/extends_iface}}
public Service(ServiceIface iface, TProtocolFactory protocolFactory, IByteBufferAllocator allocator){{#extends_iface}} : base(iface, protocolFactory, allocator){{/extends_iface}}
{
this.iface = iface;
this.protocolFactory = protocolFactory;
{{^extends_iface}}
this.allocator = allocator;
{{/extends_iface}}
{{#functions}}
this.functionMap.Add("{{name}}", async (iprot, seqid) =>
{
Expand All @@ -171,13 +185,17 @@
iprot.ReadMessageEnd();
return CreateException("{{name}}", seqid, TApplicationException.ExceptionType.ProtocolError, e.Message);
}
finally
{
iprot.Dispose();
}

try
{
{{#is_oneway}}
await iface.{{name}}Async({{{argument_list_with_args}}});
// todo: __stats_{{name}}.SuccessCounter.incr() ;
return new byte[0]; // todo: cache
return Unpooled.Empty; // todo: cache
{{/is_oneway}}

{{^is_oneway}}
Expand All @@ -198,7 +216,7 @@
{{/exceptions}}
{{/exceptions.length}}

TMemoryBuffer memoryBuffer = new TMemoryBuffer();
TByteBufferTransport memoryBuffer = new TByteBufferTransport(this.allocator);
TProtocol oprot = protocolFactory.GetProtocol(memoryBuffer);

oprot.WriteMessageBegin(new TMessage("{{name}}", TMessageType.Reply, seqid));
Expand All @@ -220,9 +238,9 @@
{{/functions}}
}

public override Task<byte[]> Invoke(byte[] request)
public override Task<IByteBuffer> Invoke(IByteBuffer request)
{
TTransport inputTransport = new TMemoryInputTransport(request);
TTransport inputTransport = new TBufferInputTransport(request);
TProtocol iprot = this.protocolFactory.GetProtocol(inputTransport);

TMessage msg;
Expand All @@ -232,19 +250,19 @@
}
catch (Exception e)
{
return TaskEx.FromException<byte[]>(e);
return TaskEx.FromException<IByteBuffer>(e);
}

Func<TProtocol, int, Task<byte[]>> fn;
Func<TProtocol, int, Task<IByteBuffer>> fn;
bool success = this.functionMap.TryGetValue(msg.Name, out fn);
if (fn == null)
{
try
{
TProtocolUtil.Skip(iprot, TType.Struct);
iprot.ReadMessageEnd();
TApplicationException x = new TApplicationException(TApplicationException.ExceptionType.UnknownMethod, "Invalid method name: '"+msg.Name+"'");
TMemoryBuffer memoryBuffer = new TMemoryBuffer();
TApplicationException x = new TApplicationException(TApplicationException.ExceptionType.UnknownMethod, "Invalid method name: '" + msg.Name + "'");
TByteBufferTransport memoryBuffer = new TByteBufferTransport(this.allocator);
TProtocol oprot = this.protocolFactory.GetProtocol(memoryBuffer);
oprot.WriteMessageBegin(new TMessage(msg.Name, TMessageType.Exception, msg.SeqID));
x.Write(oprot);
Expand All @@ -254,7 +272,7 @@
}
catch (Exception e)
{
return TaskEx.FromException<byte[]>(e);
return TaskEx.FromException<IByteBuffer>(e);
}
}

Expand Down
1 change: 1 addition & 0 deletions scrooge-generator/src/main/resources/csharpgen/service_imports.mustache
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
using System.Threading.Tasks;
using Finagle.Thrift;
using DotNetty.Buffers;
2 changes: 1 addition & 1 deletion ...ge-generator/src/main/scala/com/twitter/scrooge/csharp_generator/FunctionController.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ class FunctionController(function: TFunction, generator: CsharpGenerator, ns: Op
a.sid.name
} mkString ", "
val argument_list_with_types = function.args map { a =>
generator.typeNameOption(a.fieldType) + " " + a.sid.name
generator.typeNameOption(a) + " " + a.sid.name
} mkString ", "
val argument_list_with_args = function.args map { a =>
"args." + a.sid.name.capitalize
Expand Down