Connection does not disconnect after WriteAndFlushAsync

[francisminu]

Hi,

My understanding was that, as soon as a client connects to the server and response is sent to the client, that connection is considered complete and the client will be disconnected and channel inactive. I see that the connection persists and the method ChannelInactive is never instantiated and when new clients connect, it uses the same channel to send the connection. Please let me know why this is happening so, or if my understanding is incorrect. ie., On WriteAndFlushAsync(), I thought the connection would disconnect. Is this not the case? Do I have to explicitly mention a Channel.DisconnectAsync()? (My service acts as a server and incoming connections keep coming)

Given below is my code:

    `public class PipelinesHandler<T>: ChannelHandlerAdapter
    {
    private readonly ILoggerService _loggerService;
    private readonly IMessageProcessor _messageProcessor;
    private readonly ISocketPipe _socketPipe;
    private IByteBuffer _byteBuffer;
    private readonly ITracerService _tracerService;
    private readonly TraceHelper _traceHelper;
    private Pipe _pipe;

    public PipelinesHandler(ILoggerService loggerService, IMessageProcessor messageProcessor, ISocketPipe pipe, ITracerService tracerService)
    {
        _loggerService = loggerService;
        if(_loggerService == null)
        {
            throw new ArgumentNullException($"ILoggerService object is required.");
        }
        _messageProcessor = messageProcessor;
        if (_messageProcessor == null)
        {
            throw new ArgumentNullException($"IMessageProcessor object is required.");
        }
        _socketPipe = pipe;
        if (_socketPipe == null)
        {
            throw new ArgumentNullException($"ISocketPipe object cannot be null.");
        }
        _tracerService = tracerService;
        if (_tracerService == null)
        {
            throw new ArgumentNullException($"ISocketPipe object cannot be null.");
        }
        _traceHelper = new TraceHelper(_tracerService);
    }

    public override void ChannelActive(IChannelHandlerContext context)
    {
        _loggerService.Info(new LogMessage
        {
            ClassName = "PipelinesHandler",
            MethodName = "ChannelActive",
            MessageText = $"Channel {context.Channel.Id} is Active. Current Thread: {Thread.CurrentThread.ManagedThreadId}"
        });
        _pipe = new Pipe();
        // base.ChannelActive(context);
    }

    public override async void ChannelRead(IChannelHandlerContext ctx, object msg)
    {
        
        try
        {
            var buffer = msg as IByteBuffer;
            
            if (buffer != null)
            {
                if (buffer.IsReadable())
                {
                    var bytesReceived = new byte[buffer.ReadableBytes];
                    buffer.GetBytes(0, bytesReceived);
                    _loggerService.Info(new LogMessage
                    {
                        ClassName = "PipelinesHandler",
                        MethodName = "ChannelRead",
                        MessageText = $"Message as is read from the channel {ctx.Channel.Id}: {buffer.ToString(Encoding.ASCII)}"
                    });
                    var dataFromPipe = await ProcessData(bytesReceived);
                    if (!string.IsNullOrEmpty(dataFromPipe))
                    {
                        
                            var result = await _messageProcessor.ProcessMessage(dataFromPipe);
                            if (!string.IsNullOrEmpty(result))
                            {
                                _loggerService.Info(new LogMessage { MessageText = $"Sending result to client...Result: {result}. " });
                                byte[] messageBytes = Encoding.ASCII.GetBytes(result);
                                _byteBuffer = Unpooled.Buffer(messageBytes.Length);
                                var finalData = _byteBuffer.WriteBytes(messageBytes);
                                await ctx.WriteAndFlushAsync(finalData);
                            }
                        }
                    }
                }   
            }
        }
        catch (Exception e)
        {
            _loggerService.Error(e, new LogMessage
            {
                ClassName = "PipelinesHandler",
                MethodName = "ChannelRead",
                MessageText = $"Error occurred in Connection. Error: {e.Message}"
            });
        }
    }

    public override void ChannelReadComplete(IChannelHandlerContext context)
    {
        //context.Flush();
        //base.ChannelReadComplete(context);
    }

    public override void ChannelInactive(IChannelHandlerContext context)
    {
        _loggerService.Info(new LogMessage
        {
            ClassName = "PipelinesHandler",
            MethodName = "ChannelInactive",
            MessageText = $"Channel {context.Channel.Id} is inactive. Current Thread: {Thread.CurrentThread.ManagedThreadId}"
        });
        _pipe.Writer.Complete();
        //base.ChannelInactive(context);
    }

    public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
    {
        _loggerService.Error(exception, new LogMessage
        {
            ClassName = "PipelinesHandler",
            MethodName = "ExceptionCaught",
            MessageText = $"Exception occurred in ChannelRead method of DotNetty. Exception: {exception.Message}"
        });
        context.CloseAsync();
    }

    private async Task<string> ProcessData(object msg)
    {
        try
        {
            // process incoming data and return response
        }
        catch(Exception ex)
        {
            _loggerService.Error(ex, new LogMessage { MessageText = "Error occured in ProcessData" });
            return "";
        }
    }


    #region IDisposable Members

    /// <summary>
    ///     Close and dispose.
    /// </summary>
    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    /// <summary>
    ///     Close and dispose.
    /// </summary>
    /// <param name="disposing"></param>
    protected virtual void Dispose(bool disposing)
    {
        if (disposing)
        {
            
        }
    }

    /// <summary>
    ///     Finalizer
    /// </summary>
    ~PipelinesHandler()
    {
        Dispose(false);
    }

    #endregion
}`

Thanks,