<commit_before>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Serilog;
using Thinktecture.Relay.Server.Config;
using Thinktecture.Relay.Server.OnPremise;
using Thinktecture.Relay.Server.Repository;

namespace Thinktecture.Relay.Server.Communication
{
	internal class BackendCommunication : IBackendCommunication, IDisposable
	{
		private readonly CancellationTokenSource _cts;
		private readonly CancellationToken _cancellationToken;
		private readonly IConfiguration _configuration;
		private readonly IMessageDispatcher _messageDispatcher;
		private readonly IOnPremiseConnectorCallbackFactory _requestCallbackFactory;
		private readonly ILogger _logger;
		private readonly ILinkRepository _linkRepository;

		private readonly ConcurrentDictionary<string, IOnPremiseConnectorCallback> _requestCompletedCallbacks;
		private readonly ConcurrentDictionary<string, IOnPremiseConnectionContext> _connectionContexts;

		private readonly Dictionary<string, IDisposable> _requestSubscriptions;

		private IDisposable _responseSubscription;
		private IDisposable _acknowledgeSubscription;

		public Guid OriginId { get; }

		public BackendCommunication(IConfiguration configuration, IMessageDispatcher messageDispatcher, IOnPremiseConnectorCallbackFactory requestCallbackFactory, ILogger logger, IPersistedSettings persistedSettings, ILinkRepository linkRepository)
		{
			_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
			_messageDispatcher = messageDispatcher ?? throw new ArgumentNullException(nameof(messageDispatcher));
			_requestCallbackFactory = requestCallbackFactory ?? throw new ArgumentNullException(nameof(requestCallbackFactory));
			_logger = logger;
			_linkRepository = linkRepository ?? throw new ArgumentNullException(nameof(linkRepository));
			_requestCompletedCallbacks = new ConcurrentDictionary<string, IOnPremiseConnectorCallback>(StringComparer.OrdinalIgnoreCase);
			_connectionContexts = new ConcurrentDictionary<string, IOnPremiseConnectionContext>();
			_requestSubscriptions = new Dictionary<string, IDisposable>(StringComparer.OrdinalIgnoreCase);
			_cts = new CancellationTokenSource();
			_cancellationToken = _cts.Token;
			OriginId = persistedSettings?.OriginId ?? throw new ArgumentNullException(nameof(persistedSettings));

			_logger?.Verbose("Creating backend communication. origin-id={OriginId}", OriginId);
			_logger?.Information("Backend communication is using message dispatcher {MessageDispatcherType}", messageDispatcher.GetType().Name);
		}

		public void Prepare()
		{
			_linkRepository.DeleteAllConnectionsForOrigin(OriginId);
			_responseSubscription = StartReceivingResponses();
			_acknowledgeSubscription = StartReceivingAcknowledges();
		}

		public async Task RegisterOnPremiseAsync(IOnPremiseConnectionContext onPremiseConnectionContext)
		{
			CheckDisposed();

			_logger?.Debug("Registering link. link-id={LinkId}, connection-id={ConnectionId}, user-name={UserName}, role={Role} connector-version={ConnectorVersion}, connector-assembly-version={ConnectorAssemblyVersion}",
				onPremiseConnectionContext.LinkId,
				onPremiseConnectionContext.ConnectionId,
				onPremiseConnectionContext.UserName,
				onPremiseConnectionContext.Role,
				onPremiseConnectionContext.ConnectorVersion,
				onPremiseConnectionContext.ConnectorAssemblyVersion);

			await _linkRepository.AddOrRenewActiveConnectionAsync(onPremiseConnectionContext.LinkId, OriginId, onPremiseConnectionContext.ConnectionId, onPremiseConnectionContext.ConnectorVersion, onPremiseConnectionContext.ConnectorAssemblyVersion).ConfigureAwait(false);

			lock (_requestSubscriptions)
			{
				if (!_requestSubscriptions.ContainsKey(onPremiseConnectionContext.ConnectionId))
				{
					_requestSubscriptions[onPremiseConnectionContext.ConnectionId] = _messageDispatcher.OnRequestReceived(onPremiseConnectionContext.LinkId, onPremiseConnectionContext.ConnectionId, !onPremiseConnectionContext.SupportsAck)
						.Subscribe(request => onPremiseConnectionContext.RequestAction(request, _cancellationToken));

					onPremiseConnectionContext.IsActive = true;
				}
			}

			_connectionContexts.TryAdd(onPremiseConnectionContext.ConnectionId, onPremiseConnectionContext);
		}

		public async Task UnregisterOnPremiseConnectionAsync(string connectionId)
		{
			CheckDisposed();

			await DeactivateOnPremiseConnectionAsync(connectionId);

			_logger?.Verbose("Unregistering connection. connection-id={ConnectionId}", connectionId);
			_connectionContexts.TryRemove(connectionId, out var info);
		}

		public async Task DeactivateOnPremiseConnectionAsync(string connectionId)
		{
			if (_connectionContexts.TryGetValue(connectionId, out var connectionContext))
			{
				connectionContext.IsActive = false;
			}

			_logger?.Debug("Deactivating connection. link-id={LinkId}, connection-id={ConnectionId}", connectionContext?.LinkId, connectionId);

			await _linkRepository.RemoveActiveConnectionAsync(connectionId).ConfigureAwait(false);

			IDisposable requestSubscription;
			lock (_requestSubscriptions)
			{
				if (_requestSubscriptions.TryGetValue(connectionId, out requestSubscription))
					_requestSubscriptions.Remove(connectionId);
			}

			if (requestSubscription != null)
			{
				_logger?.Debug("Disposing request subscription. link-id={LinkId}, connection-id={ConnectionId}", connectionContext?.LinkId, connectionId);
				requestSubscription.Dispose();
			}
		}

		public Task<IOnPremiseConnectorResponse> GetResponseAsync(string requestId, TimeSpan? requestTimeout = null)
		{
			CheckDisposed();

			requestTimeout = requestTimeout ?? _configuration.OnPremiseConnectorCallbackTimeout;

			_logger?.Debug("Waiting for response. request-id={RequestId}, timeout={Timeout}", requestId, requestTimeout);

			var onPremiseConnectorCallback = _requestCompletedCallbacks[requestId] = _requestCallbackFactory.Create(requestId);

			return GetOnPremiseTargetResponseAsync(onPremiseConnectorCallback, requestTimeout, _cancellationToken);
		}

		public void SendOnPremiseConnectorRequest(Guid linkId, IOnPremiseConnectorRequest request)
		{
			CheckDisposed();

			_logger?.Debug("Dispatching request. request-id={RequestId}, link-id={LinkId}", request.RequestId, linkId);

			_messageDispatcher.DispatchRequest(linkId, request);
		}

		public async Task AcknowledgeOnPremiseConnectorRequestAsync(Guid originId, string acknowledgeId, string connectionId)<commit_msg>Use local variable to allow compilation<commit_after>using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Serilog;
using Thinktecture.Relay.Server.Config;
using Thinktecture.Relay.Server.OnPremise;
using Thinktecture.Relay.Server.Repository;

namespace Thinktecture.Relay.Server.Communication
{
	internal class BackendCommunication : IBackendCommunication, IDisposable
	{
		private readonly CancellationTokenSource _cts;
		private readonly CancellationToken _cancellationToken;
		private readonly IConfiguration _configuration;
		private readonly IMessageDispatcher _messageDispatcher;
		private readonly IOnPremiseConnectorCallbackFactory _requestCallbackFactory;
		private readonly ILogger _logger;
		private readonly ILinkRepository _linkRepository;

		private readonly ConcurrentDictionary<string, IOnPremiseConnectorCallback> _requestCompletedCallbacks;
		private readonly ConcurrentDictionary<string, IOnPremiseConnectionContext> _connectionContexts;

		private readonly Dictionary<string, IDisposable> _requestSubscriptions;

		private IDisposable _responseSubscription;
		private IDisposable _acknowledgeSubscription;

		public Guid OriginId { get; }

		public BackendCommunication(IConfiguration configuration, IMessageDispatcher messageDispatcher, IOnPremiseConnectorCallbackFactory requestCallbackFactory, ILogger logger, IPersistedSettings persistedSettings, ILinkRepository linkRepository)
		{
			_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
			_messageDispatcher = messageDispatcher ?? throw new ArgumentNullException(nameof(messageDispatcher));
			_requestCallbackFactory = requestCallbackFactory ?? throw new ArgumentNullException(nameof(requestCallbackFactory));
			_logger = logger;
			_linkRepository = linkRepository ?? throw new ArgumentNullException(nameof(linkRepository));
			_requestCompletedCallbacks = new ConcurrentDictionary<string, IOnPremiseConnectorCallback>(StringComparer.OrdinalIgnoreCase);
			_connectionContexts = new ConcurrentDictionary<string, IOnPremiseConnectionContext>();
			_requestSubscriptions = new Dictionary<string, IDisposable>(StringComparer.OrdinalIgnoreCase);
			_cts = new CancellationTokenSource();
			_cancellationToken = _cts.Token;
			OriginId = persistedSettings?.OriginId ?? throw new ArgumentNullException(nameof(persistedSettings));

			_logger?.Verbose("Creating backend communication. origin-id={OriginId}", OriginId);
			_logger?.Information("Backend communication is using message dispatcher {MessageDispatcherType}", messageDispatcher.GetType().Name);
		}

		public void Prepare()
		{
			_linkRepository.DeleteAllConnectionsForOrigin(OriginId);
			_responseSubscription = StartReceivingResponses();
			_acknowledgeSubscription = StartReceivingAcknowledges();
		}

		public async Task RegisterOnPremiseAsync(IOnPremiseConnectionContext onPremiseConnectionContext)
		{
			CheckDisposed();

			_logger?.Debug("Registering link. link-id={LinkId}, connection-id={ConnectionId}, user-name={UserName}, role={Role} connector-version={ConnectorVersion}, connector-assembly-version={ConnectorAssemblyVersion}",
				onPremiseConnectionContext.LinkId,
				onPremiseConnectionContext.ConnectionId,
				onPremiseConnectionContext.UserName,
				onPremiseConnectionContext.Role,
				onPremiseConnectionContext.ConnectorVersion,
				onPremiseConnectionContext.ConnectorAssemblyVersion);

			await _linkRepository.AddOrRenewActiveConnectionAsync(onPremiseConnectionContext.LinkId, OriginId, onPremiseConnectionContext.ConnectionId, onPremiseConnectionContext.ConnectorVersion, onPremiseConnectionContext.ConnectorAssemblyVersion).ConfigureAwait(false);

			lock (_requestSubscriptions)
			{
				if (!_requestSubscriptions.ContainsKey(onPremiseConnectionContext.ConnectionId))
				{
					_requestSubscriptions[onPremiseConnectionContext.ConnectionId] = _messageDispatcher.OnRequestReceived(onPremiseConnectionContext.LinkId, onPremiseConnectionContext.ConnectionId, !onPremiseConnectionContext.SupportsAck)
						.Subscribe(request => onPremiseConnectionContext.RequestAction(request, _cancellationToken));

					onPremiseConnectionContext.IsActive = true;
				}
			}

			_connectionContexts.TryAdd(onPremiseConnectionContext.ConnectionId, onPremiseConnectionContext);
		}

		public async Task UnregisterOnPremiseConnectionAsync(string connectionId)
		{
			CheckDisposed();

			await DeactivateOnPremiseConnectionAsync(connectionId);

			_logger?.Verbose("Unregistering connection. connection-id={ConnectionId}", connectionId);
			_connectionContexts.TryRemove(connectionId, out var info);
		}

		public async Task DeactivateOnPremiseConnectionAsync(string connectionId)
		{
			if (_connectionContexts.TryGetValue(connectionId, out var connectionContext))
			{
				connectionContext.IsActive = false;
			}

			_logger?.Debug("Deactivating connection. link-id={LinkId}, connection-id={ConnectionId}", connectionContext?.LinkId, connectionId);

			await _linkRepository.RemoveActiveConnectionAsync(connectionId).ConfigureAwait(false);

			IDisposable requestSubscription;
			lock (_requestSubscriptions)
			{
				if (_requestSubscriptions.TryGetValue(connectionId, out requestSubscription))
					_requestSubscriptions.Remove(connectionId);
			}

			if (requestSubscription != null)
			{
				_logger?.Debug("Disposing request subscription. link-id={LinkId}, connection-id={ConnectionId}", connectionContext?.LinkId, connectionId);
				requestSubscription.Dispose();
			}
		}

		public Task<IOnPremiseConnectorResponse> GetResponseAsync(string requestId, TimeSpan? requestTimeout = null)
		{
			CheckDisposed();

			var timeout = requestTimeout ?? _configuration.OnPremiseConnectorCallbackTimeout;

			_logger?.Debug("Waiting for response. request-id={RequestId}, timeout={Timeout}", requestId, timeout);

			var onPremiseConnectorCallback = _requestCompletedCallbacks[requestId] = _requestCallbackFactory.Create(requestId);

			return GetOnPremiseTargetResponseAsync(onPremiseConnectorCallback, timeout, _cancellationToken);
		}

		public void SendOnPremiseConnectorRequest(Guid linkId, IOnPremiseConnectorRequest request)
		{
			CheckDisposed();

			_logger?.Debug("Dispatching request. request-id={RequestId}, link-id={LinkId}", request.RequestId, linkId);

			_messageDispatcher.DispatchRequest(linkId, request);
		}

		public async Task AcknowledgeOnPremiseConnectorRequestAsync(Guid originId, string acknowledgeId, string connectionId)
{'fraction_non_alphanumeric': 0.07709369024856597, 'fraction_numerical': 0.0, 'mean_word_length': 7.419832582099163, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6646536', 'n_tokens_mistral': 3802, 'n_tokens_neox': 3320, 'n_words': 615}