| | | 1 | | // Licensed to the .NET Foundation under one or more agreements. |
| | | 2 | | // The .NET Foundation licenses this file to you under the MIT license. |
| | | 3 | | |
| | | 4 | | using System; |
| | | 5 | | using System.Linq; |
| | | 6 | | using System.ServiceModel; |
| | | 7 | | using System.ServiceModel.Channels; |
| | | 8 | | using System.Threading.Tasks; |
| | | 9 | | using CoreWCF.Runtime; |
| | | 10 | | |
| | | 11 | | namespace CoreWCF.ServiceModel.Channels |
| | | 12 | | { |
| | | 13 | | internal class KafkaChannelFactory : ChannelFactoryBase<IOutputChannel> |
| | | 14 | | { |
| | | 15 | | private readonly BufferManager _bufferManager; |
| | | 16 | | private readonly KafkaTransportBindingElement _transportTransportBindingElement; |
| | | 17 | | |
| | | 18 | | public KafkaChannelFactory(KafkaTransportBindingElement transportBindingElement, |
| | | 19 | | BindingContext context) |
| | 38 | 20 | | : base(context.Binding) |
| | | 21 | | { |
| | 38 | 22 | | _transportTransportBindingElement = transportBindingElement; |
| | 38 | 23 | | _bufferManager = BufferManager.CreateBufferManager(transportBindingElement.MaxBufferPoolSize, int.MaxValue); |
| | | 24 | | |
| | 38 | 25 | | MessageEncoderFactory = context.BindingParameters.Find<MessageEncodingBindingElement>()?.CreateMessageEncode |
| | 38 | 26 | | ?? KafkaConstants.DefaultMessageEncoderFactory; |
| | 38 | 27 | | } |
| | | 28 | | |
| | 944 | 29 | | public BufferManager BufferManager => _bufferManager; |
| | | 30 | | |
| | 38 | 31 | | public MessageEncoderFactory MessageEncoderFactory { get; } |
| | | 32 | | |
| | 0 | 33 | | private Task OnOpenAsync => Task.CompletedTask; |
| | | 34 | | |
| | | 35 | | protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state) |
| | | 36 | | { |
| | 0 | 37 | | return OnOpenAsync.ToApm(callback, state); |
| | | 38 | | } |
| | | 39 | | |
| | | 40 | | protected override void OnEndOpen(IAsyncResult result) |
| | | 41 | | { |
| | 0 | 42 | | result.ToApmEnd(); |
| | 0 | 43 | | } |
| | | 44 | | |
| | | 45 | | protected override void OnOpen(TimeSpan timeout) |
| | | 46 | | { |
| | | 47 | | |
| | 38 | 48 | | } |
| | | 49 | | |
| | | 50 | | protected override IOutputChannel OnCreateChannel(EndpointAddress address, Uri via) |
| | | 51 | | { |
| | 38 | 52 | | return new KafkaOutputChannel(this, address, via, MessageEncoderFactory.Encoder, _transportTransportBindingE |
| | | 53 | | } |
| | | 54 | | |
| | | 55 | | protected override void OnClosed() |
| | | 56 | | { |
| | 11 | 57 | | base.OnClosed(); |
| | 11 | 58 | | _bufferManager.Clear(); |
| | 11 | 59 | | } |
| | | 60 | | } |
| | | 61 | | } |