<commit_before>﻿using System;
using System.Threading.Tasks;
using Esb.Cluster;
using Esb.Cluster.Proccessors;
using Esb.Message;
using Esb.Transport;

namespace Esb.Processing
{
    /// <summary>
    /// The local worker process; has a own node with configuration
    /// </summary>
    public class Worker : IWorker
    {
        public INodeConfiguration LocalNode { get; private set; }
        
        private IMessageQueue _messageQueue;
        private readonly IRouter _router;

        private readonly IClusterConfiguration _clusterConfiguration;
        private readonly WorkerConfiguration _workerConfiguration;

        public Worker(WorkerConfiguration workerConfiguration, IRouter router, IMessageQueue messageQueue)
        {
            _workerConfiguration = workerConfiguration;
            _router = router;
            _messageQueue = messageQueue;
            _clusterConfiguration = new ClusterConfiguration();
            
            CreateLocalNodeConfiguration();
            AddClusterCommunicationProcessors();
            InitialStartUpAync();
        }

        public WorkerStatus Status { get; private set; } = WorkerStatus.Initialization;

        public void InitialStartUpAync()
        {
            Task.Factory.StartNew(() =>
            {
                Start();
            });
        }

        private void FindClusterAndEstablishCommunication()
        {
            if (_workerConfiguration.ControllerNodes.Empty() && IsController)
                return;

            foreach (var controllerNode in _workerConfiguration.ControllerNodes)
            {
                if (_router.ProcessSync(new Envelope(new AskForClusterConfiguration(), Priority.Administrative),
                    new NodeConfiguration(this, Guid.Empty, controllerNode)))
                    break;
            }
        }

        private void AddClusterCommunicationProcessors()
        {
            LocalNode.Processors.Add(new RemoveNodeFromCluster());
            LocalNode.Processors.Add(new AddNodeToCluster());
            LocalNode.Processors.Add(new SyncClusterConfiguration());

            if (!IsController)
                return;

            LocalNode.Processors.Add(new AskForClusterConfiguration());
        }

        private void CreateLocalNodeConfiguration()
        {
            LocalNode = new NodeConfiguration(this, _workerConfiguration.NodeId, _workerConfiguration.Address);
        }

        public void Start()
        {
            Status = WorkerStatus.Initialization;
            FindClusterAndEstablishCommunication();
            Status = WorkerStatus.Starting;
            SetLocalNodeOnline();
            SendOnlineMessage();
            Status = WorkerStatus.Started;
        }

        private void SetLocalNodeOnline()
        {
            _clusterConfiguration.AddNode(LocalNode);
        }

        private void SendOnlineMessage()
        {
            _router.Process(new Envelope(new Cluster.Messages.AddNodeToCluster(LocalNode), Priority.Administrative));
        }

        public void Stop()
        {
            Status = WorkerStatus.Stopping;
            SendOfflineMessage();
            SetLocalNodeOffline();
            Status = WorkerStatus.Stopped;
        }<commit_msg>some constraints for starting and stopping the worker<commit_after>﻿using System;
using System.Threading.Tasks;
using Esb.Cluster;
using Esb.Cluster.Proccessors;
using Esb.Message;
using Esb.Transport;

namespace Esb.Processing
{
    /// <summary>
    /// The local worker process; has a own node with configuration
    /// </summary>
    public class Worker : IWorker
    {
        public INodeConfiguration LocalNode { get; private set; }
        
        private IMessageQueue _messageQueue;
        private readonly IRouter _router;

        private readonly IClusterConfiguration _clusterConfiguration;
        private readonly WorkerConfiguration _workerConfiguration;

        public Worker(WorkerConfiguration workerConfiguration, IRouter router, IMessageQueue messageQueue)
        {
            _workerConfiguration = workerConfiguration;
            _router = router;
            _messageQueue = messageQueue;
            _clusterConfiguration = new ClusterConfiguration();
            
            CreateLocalNodeConfiguration();
            AddClusterCommunicationProcessors();
            InitialStartUpAync();
        }

        public WorkerStatus Status { get; private set; } = WorkerStatus.Stopped;

        public void InitialStartUpAync()
        {
            Task.Factory.StartNew(() =>
            {
                Start();
            });
        }

        private void FindClusterAndEstablishCommunication()
        {
            if (_workerConfiguration.ControllerNodes.Empty() && IsController)
                return;

            foreach (var controllerNode in _workerConfiguration.ControllerNodes)
            {
                if (_router.ProcessSync(new Envelope(new AskForClusterConfiguration(), Priority.Administrative),
                    new NodeConfiguration(this, Guid.Empty, controllerNode)))
                    break;
            }
        }

        private void AddClusterCommunicationProcessors()
        {
            LocalNode.Processors.Add(new RemoveNodeFromCluster());
            LocalNode.Processors.Add(new AddNodeToCluster());
            LocalNode.Processors.Add(new SyncClusterConfiguration());

            if (!IsController)
                return;

            LocalNode.Processors.Add(new AskForClusterConfiguration());
        }

        private void CreateLocalNodeConfiguration()
        {
            LocalNode = new NodeConfiguration(this, _workerConfiguration.NodeId, _workerConfiguration.Address);
        }

        public void Start()
        {
            if (Status != WorkerStatus.Stopped)
                throw new Exception("Cannot start worker, because it is not stopped");
            Status = WorkerStatus.Initialization;
            FindClusterAndEstablishCommunication();
            Status = WorkerStatus.Starting;
            SetLocalNodeOnline();
            SendOnlineMessage();
            Status = WorkerStatus.Started;
        }

        private void SetLocalNodeOnline()
        {
            _clusterConfiguration.AddNode(LocalNode);
        }

        private void SendOnlineMessage()
        {
            _router.Process(new Envelope(new Cluster.Messages.AddNodeToCluster(LocalNode), Priority.Administrative));
        }

        public void Stop()
        {
            if (Status != WorkerStatus.Starting)
                throw new Exception("Cannot stop worker, because it is not started");

            Status = WorkerStatus.Stopping;
            SendOfflineMessage();
            SetLocalNodeOffline();
            Status = WorkerStatus.Stopped;
        }
{'fraction_non_alphanumeric': 0.0738235294117647, 'fraction_numerical': 0.0, 'mean_word_length': 2.0759837177747626, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '7186244', 'n_tokens_mistral': 1585, 'n_tokens_neox': 1513, 'n_words': 349}