diff options
Diffstat (limited to 'lib/netstd/Thrift/Server/TSimpleAsyncServer.cs')
| -rw-r--r-- | lib/netstd/Thrift/Server/TSimpleAsyncServer.cs | 196 |
1 files changed, 196 insertions, 0 deletions
diff --git a/lib/netstd/Thrift/Server/TSimpleAsyncServer.cs b/lib/netstd/Thrift/Server/TSimpleAsyncServer.cs new file mode 100644 index 000000000..a0a3e4c15 --- /dev/null +++ b/lib/netstd/Thrift/Server/TSimpleAsyncServer.cs @@ -0,0 +1,196 @@ +// 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.Protocol; +using Thrift.Processor; +using Thrift.Transport; + +namespace Thrift.Server +{ + //TODO: unhandled exceptions, etc. + + // ReSharper disable once InconsistentNaming + public class TSimpleAsyncServer : TServer + { + private readonly int _clientWaitingDelay; + private volatile Task _serverTask; + + public TSimpleAsyncServer(ITAsyncProcessor processor, TServerTransport serverTransport, + ITProtocolFactory inputProtocolFactory, ITProtocolFactory outputProtocolFactory, + ILoggerFactory loggerFactory, int clientWaitingDelay = 10) + : this(new TSingletonProcessorFactory(processor), serverTransport, + new TTransportFactory(), new TTransportFactory(), + inputProtocolFactory, outputProtocolFactory, + loggerFactory.CreateLogger(nameof(TSimpleAsyncServer)), clientWaitingDelay) + { + } + + public TSimpleAsyncServer(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(TTransport client, CancellationToken cancellationToken) + { + Logger.LogTrace("Started client request processing"); + + var processor = ProcessorFactory.GetAsyncProcessor(client, this); + + TTransport inputTransport = null; + TTransport outputTransport = null; + TProtocol inputProtocol = null; + TProtocol outputProtocol = null; + object connectionContext = null; + + try + { + 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); + } + + } + finally + { + //Close transports + inputTransport?.Close(); + outputTransport?.Close(); + + // disposable stuff should be disposed + inputProtocol?.Dispose(); + outputProtocol?.Dispose(); + inputTransport?.Dispose(); + outputTransport?.Dispose(); + } + + Logger.LogTrace("Completed client request processing"); + } + } +} |