﻿/*
 * Licensed 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.
 */

namespace DotPulsar.Internal
{
    using Abstractions;
    using Extensions;
    using PulsarApi;
    using System;
    using System.Buffers;
    using System.Threading.Tasks;

    public sealed class ChannelManager : IDisposable
    {
        private readonly RequestResponseHandler _requestResponseHandler;
        private readonly IdLookup<IChannel> _consumerChannels;
        private readonly IdLookup<IChannel> _producerChannels;
        private readonly EnumLookup<BaseCommand.Type, Action<BaseCommand>> _incoming;

        public ChannelManager()
        {
            _requestResponseHandler = new RequestResponseHandler();
            _consumerChannels = new IdLookup<IChannel>();
            _producerChannels = new IdLookup<IChannel>();
            _incoming = new EnumLookup<BaseCommand.Type, Action<BaseCommand>>(cmd => _requestResponseHandler.Incoming(cmd));
            _incoming.Set(BaseCommand.Type.CloseConsumer, cmd => Incoming(cmd.CloseConsumer));
            _incoming.Set(BaseCommand.Type.CloseProducer, cmd => Incoming(cmd.CloseProducer));
            _incoming.Set(BaseCommand.Type.ActiveConsumerChange, cmd => Incoming(cmd.ActiveConsumerChange));
            _incoming.Set(BaseCommand.Type.ReachedEndOfTopic, cmd => Incoming(cmd.ReachedEndOfTopic));
        }

        public bool HasChannels()
            => !_consumerChannels.IsEmpty() || !_producerChannels.IsEmpty();

        public Task<ProducerResponse> Outgoing(CommandProducer command, IChannel channel)
        {
            var producerId = _producerChannels.Add(channel);
            command.ProducerId = producerId;
            var response = _requestResponseHandler.Outgoing(command);

            return response.ContinueWith(result =>
            {
                if (result.Result.CommandType == BaseCommand.Type.Error)
                {
                    _ = _producerChannels.Remove(producerId);
                    result.Result.Error.Throw();
                }

                channel.Connected();

                return new ProducerResponse(producerId, result.Result.ProducerSuccess.ProducerName);
            }, TaskContinuationOptions.OnlyOnRanToCompletion);
        }

        public Task<SubscribeResponse> Outgoing(CommandSubscribe command, IChannel channel)
        {
            var consumerId = _consumerChannels.Add(channel);
            command.ConsumerId = consumerId;
            var response = _requestResponseHandler.Outgoing(command);

            return response.ContinueWith(result =>
            {
                if (result.Result.CommandType == BaseCommand.Type.Error)
                {
                    _ = _consumerChannels.Remove(consumerId);
                    result.Result.Error.Throw();
                }

                channel.Connected();

                return new SubscribeResponse(consumerId);
            }, TaskContinuationOptions.OnlyOnRanToCompletion);
        }

        public Task<BaseCommand> Outgoing(CommandCloseConsumer command)
        {
            var consumerId = command.ConsumerId;

            Task<BaseCommand> response;

            using (TakeConsumerSenderLock(consumerId))
            {
                response = _requestResponseHandler.Outgoing(command);
            }

            _ = response.ContinueWith(result =>
            {
                if (result.Result.CommandType == BaseCommand.Type.Success)
                    _ = _consumerChannels.Remove(consumerId);
            }, TaskContinuationOptions.OnlyOnRanToCompletion);

            return response;
        }

        public Task<BaseCommand> Outgoing(CommandCloseProducer command)
        {
            var producerId = command.ProducerId;

            Task<BaseCommand> response;

            using (TakeProducerSenderLock(producerId))
            {
                response = _requestResponseHandler.Outgoing(command);
            }

            _ = response.ContinueWith(result =>
            {
                if (result.Result.CommandType == BaseCommand.Type.Success)
                    _ = _producerChannels.Remove(producerId);
            }, TaskContinuationOptions.OnlyOnRanToCompletion);

            return response;
        }

        public Task<BaseCommand> Outgoing(CommandUnsubscribe command)
        {
            var consumerId = command.ConsumerId;

            Task<BaseCommand> response;

            using (TakeConsumerSenderLock(consumerId))
            {
                response = _requestResponseHandler.Outgoing(command);
            }

            _ = response.ContinueWith(result =>
            {
                if (result.Result.CommandType == BaseCommand.Type.Success)
                   _consumerChannels.Remove(consumerId)?.Unsubscribed();
            }, TaskContinuationOptions.OnlyOnRanToCompletion);

            return response;
        }

        public Task<BaseCommand> Outgoing(CommandSend command)
        {
            using (TakeProducerSenderLock(command.ProducerId))
            {
                return _requestResponseHandler.Outgoing(command);
            }
        }

        public Task<BaseCommand> Outgoing(CommandGetOrCreateSchema command)
            => _requestResponseHandler.Outgoing(command);

        public Task<BaseCommand> Outgoing(CommandConnect command)
            => _requestResponseHandler.Outgoing(command);

        public Task<BaseCommand> Outgoing(CommandLookupTopic command)
            => _requestResponseHandler.Outgoing(command);

        public Task<BaseCommand> Outgoing(CommandPartitionedTopicMetadata command)
            => _requestResponseHandler.Outgoing(command);

        public Task<BaseCommand> Outgoing(CommandSeek command)
        {
            using (TakeConsumerSenderLock(command.ConsumerId))
            {
                return _requestResponseHandler.Outgoing(command);
            }
        }

        public Task<BaseCommand> Outgoing(CommandGetLastMessageId command)
        {
            using (TakeConsumerSenderLock(command.ConsumerId))
            {
                return _requestResponseHandler.Outgoing(command);
            }
        }

        public void Incoming(BaseCommand command)
            => _incoming.Get(command.CommandType)(command);

        public void Incoming(CommandMessage command, ReadOnlySequence<byte> data)
            => _consumerChannels[command.ConsumerId]?.Received(new MessagePackage(command.MessageId, command.RedeliveryCount, data));

        public void Dispose()
        {
            _requestResponseHandler.Dispose();

            foreach (var channel in _consumerChannels.RemoveAll())
                channel.Disconnected();

            foreach (var channel in _producerChannels.RemoveAll())
                channel.Disconnected();
        }

        private void Incoming(CommandReachedEndOfTopic command)
            => _consumerChannels[command.ConsumerId]?.ReachedEndOfTopic();

        private void Incoming(CommandCloseConsumer command)
        {
            var channel = _consumerChannels[command.ConsumerId];

            if (channel is null)
                return;

            _ = _consumerChannels.Remove(command.ConsumerId);
            _requestResponseHandler.Incoming(command);
            channel.ClosedByServer();
        }

        private void Incoming(CommandCloseProducer command)
        {
            var channel = _producerChannels[command.ProducerId];

            if (channel is null)
                return;

            _ = _producerChannels.Remove(command.ProducerId);
            _requestResponseHandler.Incoming(command);
            channel.ClosedByServer();
        }

        private void Incoming(CommandActiveConsumerChange command)
        {
            var channel = _consumerChannels[command.ConsumerId];

            if (channel is null)
                return;

            if (command.IsActive)
                channel.Activated();
            else
                channel.Deactivated();
        }

        private IDisposable TakeConsumerSenderLock(ulong consumerId)
        {
            var channel = _consumerChannels[consumerId];
            if (channel is null)
                throw new OperationCanceledException();

            return channel.SenderLock();
        }

        private IDisposable TakeProducerSenderLock(ulong producerId)
        {
            var channel = _producerChannels[producerId];
            if (channel is null)
                throw new OperationCanceledException();

            return channel.SenderLock();
        }
    }
}