/// <summary>
        /// Performs configuration
        /// </summary>
        /// <param name="configuration">Previous configuration</param>
        /// <param name="config">Akka configuration</param>
        /// <returns>Updated configuration</returns>
        public LoggerConfiguration Configure(LoggerConfiguration configuration, Config config)
        {
            var minimumLevel = config.GetString("ClusterKit.Log.ElasticSearch.minimumLevel", "none")?.Trim();

            LogEventLevel level;
            if (!Enum.TryParse(minimumLevel, true, out level))
            {
                return configuration;
            }

            var nodes = config.GetStringList("ClusterKit.Log.ElasticSearch.nodes");

            var indexFormat = config.GetString("ClusterKit.Log.ElasticSearch.indexFormat", "logstash-{0:yyyy.MM.dd}");

            Log.Information(
                "{Type}: \n\tMinimum level: {MinimumLevel}\n\tIndex format: {IndexFormat}\n\tNodes:\n\t\t{NodeList}\n",
                this.GetType().FullName,
                minimumLevel,
                indexFormat,
                string.Join("\n\t\t", nodes));


            SelfLog.Enable(Console.WriteLine);
            var options = new ElasticsearchSinkOptions(nodes.Select(s => new Uri(s)))
                              {
                                  MinimumLogEventLevel = level,
                                  AutoRegisterTemplate = true,
                                  IndexFormat = indexFormat
                              };

            return configuration.WriteTo.Elasticsearch(options);


        }

 public RemoteSettings(Config config)
 {
     Config = config;
     LogReceive = config.GetBoolean("akka.remote.log-received-messages");
     LogSend = config.GetBoolean("akka.remote.log-sent-messages");
     UntrustedMode = config.GetBoolean("akka.remote.untrusted-mode");
     TrustedSelectionPaths = new HashSet<string>(config.GetStringList("akka.remote.trusted-selection-paths"));
     RemoteLifecycleEventsLogLevel = config.GetString("akka.remote.log-remote-lifecycle-events") ?? "DEBUG";
     if (RemoteLifecycleEventsLogLevel.Equals("on")) RemoteLifecycleEventsLogLevel = "DEBUG";
     FlushWait = config.GetMillisDuration("akka.remote.flush-wait-on-shutdown");
     ShutdownTimeout = config.GetMillisDuration("akka.remote.shutdown-timeout");
     TransportNames = config.GetStringList("akka.remote.enabled-transports");
     Transports = (from transportName in TransportNames
         let transportConfig = TransportConfigFor(transportName)
         select new TransportSettings(transportConfig)).ToArray();
     Adapters = ConfigToMap(config.GetConfig("akka.remote.adapters"));
     BackoffPeriod = config.GetMillisDuration("akka.remote.backoff-interval");
     RetryGateClosedFor = config.GetMillisDuration("akka.remote.retry-gate-closed-for", TimeSpan.Zero);
     UsePassiveConnections = config.GetBoolean("akka.remote.use-passive-connections");
     SysMsgBufferSize = config.GetInt("akka.remote.system-message-buffer-size");
     SysResendTimeout = config.GetMillisDuration("akka.remote.resend-interval");
     InitialSysMsgDeliveryTimeout = config.GetMillisDuration("akka.remote.initial-system-message-delivery-timeout");
     SysMsgAckTimeout = config.GetMillisDuration("akka.remote.system-message-ack-piggyback-timeout");
     QuarantineDuration = config.GetMillisDuration("akka.remote.prune-quarantine-marker-after");
     StartupTimeout = config.GetMillisDuration("akka.remote.startup-timeout");
     CommandAckTimeout = config.GetMillisDuration("akka.remote.command-ack-timeout");
 }

 public StringPathEntry(bool valid, bool exists, Config config, string value) : this()
 {
     Config = config;
     Exists = exists;
     Valid = valid;
     Value = value;
 }

        public CurrentSynchronizationContextDispatcherConfigurator(Config config, IDispatcherPrerequisites prerequisites)
            : base(config, prerequisites)
        {

            _executorServiceConfigurator = new CurrentSynchronizationContextExecutorServiceFactory(config, prerequisites);
            // We don't bother trying to support any other type of exectuor here. PinnedDispatcher doesn't support them
        }

        static PostgreSqlSnapshotStoreSpec()
        {
            var connectionString = ConfigurationManager.ConnectionStrings["TestDb"].ConnectionString;

            var config = @"
                akka.persistence {
                    publish-plugin-commands = on
                    snapshot-store {
                        plugin = ""akka.persistence.snapshot-store.postgresql""
                        postgresql {
                            class = ""Akka.Persistence.PostgreSql.Snapshot.PostgreSqlSnapshotStore, Akka.Persistence.PostgreSql""
                            plugin-dispatcher = ""akka.actor.default-dispatcher""
                            table-name = snapshot_store
                            schema-name = public
                            auto-initialize = on
                            connection-string = """ + connectionString + @"""
                        }
                    }
                }";

            SpecConfig = ConfigurationFactory.ParseString(config);

            //need to make sure db is created before the tests start
            DbUtils.Initialize();
        }

 private static ActorSystem CreateClient(Config commonConfig)
 {
     var config = commonConfig.WithFallback("akka.remote.helios.tcp.port = 9002");
     var system = ActorSystem.Create("Client", config);
     DeadRequestProcessingActor.Install(system);
     return system;
 }

 protected PersistenceSpec(Config config = null, ITestOutputHelper output = null)
     : base(config, output)
 {
     _name = NamePrefix + "-" + _counter.GetAndIncrement();
     Clean = new Cleanup(this);
     Clean.Initialize();
 }

 public Deployer(Settings settings)
 {
     _settings = settings;
     _deployment = settings.Config.GetConfig("akka.actor.deployment");
     _default = _deployment.GetConfig("default");
     Init();
 }

        private TestKitBase(TestKitAssertions assertions, ActorSystem system, Config config, string actorSystemName = null)
        {
            if(assertions == null) throw new ArgumentNullException("assertions");
            if(system == null)
            {
                var configWithDefaultFallback = config.SafeWithFallback(_defaultConfig);
                system = ActorSystem.Create(actorSystemName ?? "test", configWithDefaultFallback);
            }

            _assertions = assertions;
            _system = system;
            system.RegisterExtension(new TestKitExtension());
            system.RegisterExtension(new TestKitAssertionsExtension(assertions));
            _testKitSettings = TestKitExtension.For(_system);
            _queue = new BlockingQueue<MessageEnvelope>();
            _log = Logging.GetLogger(system, GetType());


            var testActor = CreateTestActor(system, "testActor" + _testActorId.IncrementAndGet());
            _testActor = testActor;
            //Wait for the testactor to start
            AwaitCondition(() =>
            {
                var repRef = _testActor as RepointableRef;
                return repRef == null || repRef.IsStarted;
            }, TimeSpan.FromSeconds(1), TimeSpan.FromMilliseconds(10));
        }

 protected JournalSpec(Config config = null, string actorSystemName = null, string testActorName = null)
     : base(config ?? Config, actorSystemName ?? "JournalSpec", testActorName)
 {
     _senderProbe = CreateTestProbe();
     _receiverProbe = CreateTestProbe();
     WriteMessages(1, 5, Pid, _senderProbe.Ref);
 }

 /// <summary>
 /// Performs configuration
 /// </summary>
 /// <param name="configuration">Previous configuration</param>
 /// <param name="config">Akka configuration</param>
 /// <returns>Updated configuration</returns>
 public LoggerConfiguration Configure(LoggerConfiguration configuration, Config config)
 {
     var templateName = config.GetString("ClusterKit.NodeManager.NodeTemplate");
     return string.IsNullOrWhiteSpace(templateName)
                ? configuration
                : configuration.Enrich.WithProperty("nodeTemplate", templateName);
 }

        static SqlServerJournalSpec()
        {
            var connectionString = ConfigurationManager.ConnectionStrings["TestDb"].ConnectionString.Replace(@"\", "\\");

            var specString = @"
                    akka.persistence {
                        publish-plugin-commands = on
                        journal {
                            plugin = ""akka.persistence.journal.sql-server""
                            sql-server {
                                class = ""Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer""
                                plugin-dispatcher = ""akka.actor.default-dispatcher""
                                table-name = EventJournal
                                schema-name = dbo
                                auto-initialize = on
                                connection-string = ""Data Source=localhost\\SQLEXPRESS;Database=akka_persistence_tests;User Id=akkadotnet;Password=akkadotnet;""
                            }
                        }
                    }";

            SpecConfig = ConfigurationFactory.ParseString(specString);


            //need to make sure db is created before the tests start
            DbUtils.Initialize();
        }

 public JournalSettings(Config config)
 {
     if (config == null) throw new ArgumentNullException("config", "Table Storage journal settings cannot be initialized, because required HOCON section couldn't be found");
     TableName = config.GetString("table-name");
     ConnectionStrings = config.GetStringList("connection-strings");
     _settings = new AzureStorageSettings(ConnectionStrings);
 }

        private TestKitBase(TestKitAssertions assertions, ActorSystem system, Config config, string actorSystemName, string testActorName)
        {
            if(assertions == null) throw new ArgumentNullException("assertions");
            if(system == null)
            {
                var configWithDefaultFallback = config.SafeWithFallback(_defaultConfig);
                system = ActorSystem.Create(actorSystemName ?? "test", configWithDefaultFallback);
            }

            _assertions = assertions;
            _system = system;
            system.RegisterExtension(new TestKitExtension());
            system.RegisterExtension(new TestKitAssertionsExtension(assertions));
            _testKitSettings = TestKitExtension.For(_system);
            _queue = new BlockingQueue<MessageEnvelope>();
            _log = Logging.GetLogger(system, GetType());
            _eventFilterFactory = new EventFilterFactory(this);
            if (string.IsNullOrEmpty(testActorName))
                testActorName = "testActor" + _testActorId.IncrementAndGet();

            var testActor = CreateTestActor(system, testActorName);
            //Wait for the testactor to start
            AwaitCondition(() =>
            {
                var repRef = testActor as RepointableRef;
                return repRef == null || repRef.IsStarted;
            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));

            if(!(this is NoImplicitSender))
            {
                InternalCurrentActorCellKeeper.Current = (ActorCell)((ActorRefWithCell)testActor).Underlying;
            }
            _testActor = testActor;

        }

        public override Deploy ParseConfig(string key, Config config)
        {
            var deploy = base.ParseConfig(key, config);
            if (deploy == null) return null;

            if (deploy.Config.GetBoolean("cluster.enabled"))
            {
                if(deploy.Scope != Deploy.NoScopeGiven)
                    throw new ConfigurationException(string.Format("Cluster deployment can't be combined with scope [{0}]", deploy.Scope));
                //TODO: add handling for RemoteRouterConfig

                if (deploy.RouterConfig is Pool)
                {
                    return
                        deploy.Copy(scope: ClusterScope.Instance)
                            .WithRouterConfig(new ClusterRouterPool(deploy.RouterConfig as Pool,
                                ClusterRouterPoolSettings.FromConfig(deploy.Config)));
                }
                else if (deploy.RouterConfig is Group)
                {
                    return
                        deploy.Copy(scope: ClusterScope.Instance)
                            .WithRouterConfig(new ClusterRouterGroup(deploy.RouterConfig as Group,
                                ClusterRouterGroupSettings.FromConfig(deploy.Config)));
                }
                else
                {
                    throw new ArgumentException(string.Format("Cluster-aware router can only wrap Pool or Group, got [{0}]", deploy.RouterConfig.GetType()));
                }
            }
            else
            {
                return deploy;
            }
        }

 internal static TimeSpan? GetSafeDeadlockTimeout(Config cfg)
 {
     var timespan = cfg.GetTimeSpan("deadlock-timeout", TimeSpan.FromSeconds(-1));
     if (timespan.TotalSeconds < 0)
         return null;
     return timespan;
 }

        static OracleSnapshotStoreSpec()
        {
            var specString = @"
                        akka.persistence {
                            publish-plugin-commands = on
                            snapshot-store {
                                plugin = ""akka.persistence.snapshot-store.oracle-managed""
                                oracle-managed {
                                    class = ""Akka.Persistence.OracleManaged.Snapshot.OracleSnapshotStore, Akka.Persistence.OracleManaged""
                                    plugin-dispatcher = ""akka.actor.default-dispatcher""
                                    table-name = Spec-SnapshotStore
                                    schema-name = akka_persist_tests
                                    auto-initialize = on
                                    connection-string-name = ""TestDb""
                                }
                            }
                        }";

            SpecConfig = ConfigurationFactory.ParseString(specString);


            //need to make sure db is created before the tests start
            //DbUtils.Initialize();
            DbUtils.Clean("Spec-SnapshotStore");
        }

        /// <summary>
        /// Creates cluster publish subscribe settings from provided configuration with the same layout as `akka.cluster.pub-sub`.
        /// </summary>
        public static DistributedPubSubSettings Create(Config config)
        {
            RoutingLogic routingLogic = null;
            var routingLogicName = config.GetString("routing-logic");
            switch (routingLogicName)
            {
                case "random":
                    routingLogic = new RandomLogic();
                    break;
                case "round-robin":
                    routingLogic = new RoundRobinRoutingLogic();
                    break;
                case "broadcast":
                    routingLogic = new BroadcastRoutingLogic();
                    break;
                case "consistent-hashing":
                    throw new ArgumentException("Consistent hashing routing logic cannot be used by the pub-sub mediator");
                default:
                    throw new ArgumentException("Unknown routing logic is tried to be applied to the pub-sub mediator: " +
                                                routingLogicName);
            }

            return new DistributedPubSubSettings(
                config.GetString("role"),
                routingLogic,
                config.GetTimeSpan("gossip-interval"),
                config.GetTimeSpan("removed-time-to-live"),
                config.GetInt("max-delta-elements"));
        }

        private static void Main(string[] args)
        {
            var section = (AkkaConfigurationSection)ConfigurationManager.GetSection("akka");
            _clusterConfig = section.AkkaConfig;
            LaunchBackend(new[] { "2551" });
            LaunchBackend(new[] { "2552" });
            LaunchBackend(new string[0]);

            string input;

            Console.WriteLine("Enter send to send the message bar or quit to exit.");

            while ((input = Console.ReadLine()) != null)
            {
                var cmd = input;
                switch (cmd)
                {
                    case "quit":
                        return; // Stop the run thread
                    case "send":
                        // Send to backend which will write to the console
                        SendToBackend();
                        break;
                }
            }

            Console.WriteLine("Press any key to exit.");
            Console.ReadKey();
        }

        public SessionSettings(Config config)
        {
            if (config == null) throw new ArgumentNullException("config");

            Builder = Cluster.Builder();
            
            // Get IP and port configuration
            int port = config.GetInt("port", 9042);
            IPEndPoint[] contactPoints = ParseContactPoints(config.GetStringList("contact-points"), port);
            Builder.AddContactPoints(contactPoints);

            // Support user/pass authentication
            if (config.HasPath("credentials"))
                Builder.WithCredentials(config.GetString("credentials.username"), config.GetString("credentials.password"));

            // Support SSL
            if (config.GetBoolean("ssl"))
                Builder.WithSSL();

            // Support compression
            string compressionTypeConfig = config.GetString("compression");
            if (compressionTypeConfig != null)
            {
                var compressionType = (CompressionType) Enum.Parse(typeof (CompressionType), compressionTypeConfig, true);
                Builder.WithCompression(compressionType);
            }
        }