public async Task Init( string name, IProviderRuntime providerRuntime, IProviderConfiguration config )
    {
      Log = providerRuntime.GetLogger( this.GetType().Name );

      try
      {
        ConfigureJsonSerializerSettings( config );

        if( !config.Properties.ContainsKey( "DataConnectionString" ) )
        {
          throw new BadProviderConfigException(
            "The DataConnectionString setting has not been configured in the cloud role. Please add a DataConnectionString setting with a valid Azure Storage connection string." );
        }
        else
        {
          var account = CloudStorageAccount.Parse( config.Properties[ "DataConnectionString" ] );
          var blobClient = account.CreateCloudBlobClient();
          var containerName = config.Properties.ContainsKey( "ContainerName" ) ? config.Properties[ "ContainerName" ] : "grainstate";
          container = blobClient.GetContainerReference( containerName );
          await container.CreateIfNotExistsAsync();
        }
      }
      catch( Exception ex )
      {
        Log.Error( 0, ex.ToString(), ex );
        throw;
      }
    }

        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            string connectionStringName = config.Properties["ConnectionStringName"];

            if (string.IsNullOrEmpty(connectionStringName))
            {
                this.Log.Info("Starting RavenDB Storage Provider InMemory");
                return this.InMemoryMode();
            }

            var settings = ConfigurationManager.ConnectionStrings[connectionStringName];

            var connectionStringBuilder = new DbConnectionStringBuilder
            {
                ConnectionString = settings.ConnectionString
            };

            object url;
            if (connectionStringBuilder.TryGetValue("Url", out url))
            {
                this.Log.Info("Starting RavenDB Storage Provider attached to server {0}", url);
                return this.ServerMode(connectionStringName);
            }

            object dataDir;
            if (connectionStringBuilder.TryGetValue("DataDir", out dataDir))
            {
                this.Log.Info("Starting RavenDB Storage Provider embedded in directory {0}", dataDir);
                return this.LocalMode(connectionStringName);
            }

            return TaskDone.Done;
        }

        public override async Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            await base.Init(name, providerRuntime, config);

            long grainId = GrainCallBootstrapTestConstants.GrainId;
            int a = GrainCallBootstrapTestConstants.A;
            int b = GrainCallBootstrapTestConstants.B;
            ISimpleGrain grain = providerRuntime.GrainFactory.GetGrain<ISimpleGrain>(grainId, SimpleGrain.SimpleGrainNamePrefix);
            
            logger.Info("Setting A={0} on {1}", a, grainId);
            await grain.SetA(a);
            logger.Info("Setting B={0} on {1}", b, grainId);
            await grain.SetB(b);
            
            logger.Info("Getting AxB from {0}", grainId);
            int axb = await grain.GetAxB();
            logger.Info("Got AxB={0} from {1}", axb, grainId);
            
            int expected = a * b;
            int actual = axb;
            if (expected != actual) {
                throw new Exception(string.Format(
                    "Value returned to {0} by {1} should be {2} not {3}", 
                     GetType().Name, grainId, expected, actual));
            }
        }

        /// <summary> Init the factory.</summary>
        public virtual void Init(IProviderConfiguration config, string providerName, Logger logger, IServiceProvider serviceProvider)
        {
            if (config == null) throw new ArgumentNullException("config");
            if (!config.Properties.TryGetValue(DataConnectionStringPropertyName, out dataConnectionString))
                throw new ArgumentException(String.Format("{0} property not set", DataConnectionStringPropertyName));
            if (!config.Properties.TryGetValue(DeploymentIdPropertyName, out deploymentId))
                throw new ArgumentException(String.Format("{0} property not set", DeploymentIdPropertyName));

            cacheSize = SimpleQueueAdapterCache.ParseSize(config, CacheSizeDefaultValue);

            string numQueuesString;
            numQueues = NumQueuesDefaultValue;
            if (config.Properties.TryGetValue(NumQueuesPropertyName, out numQueuesString))
            {
                if (!int.TryParse(numQueuesString, out numQueues))
                    throw new ArgumentException(String.Format("{0} invalid.  Must be int", NumQueuesPropertyName));
            }

            this.providerName = providerName;
            streamQueueMapper = new HashRingBasedStreamQueueMapper(numQueues, providerName);
            adapterCache = new SimpleQueueAdapterCache(cacheSize, logger);
            if (StreamFailureHandlerFactory == null)
            {
                StreamFailureHandlerFactory =
                    qid => Task.FromResult<IStreamFailureHandler>(new NoOpStreamDeliveryFailureHandler(false));
            }
        }

        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            this.logger = providerRuntime.GetLogger("Dashboard");

            var router = new Router();
            new DashboardController(router, TaskScheduler.Current,  providerRuntime);

            var options = new StartOptions
            {
                ServerFactory = "Nowin",
                Port = config.Properties.ContainsKey("Port") ? int.Parse(config.Properties["Port"]) : 8080,
            };

            var username = config.Properties.ContainsKey("Username") ? config.Properties["Username"] : null;
            var password = config.Properties.ContainsKey("Password") ? config.Properties["Password"] : null;
            try
            {
                host = WebApp.Start(options, app => new WebServer(router, username, password).Configuration(app));
            }
            catch (Exception ex)
            {
                this.logger.Error(10001, ex.ToString());
            }

            this.logger.Verbose($"Dashboard listening on {options.Port}");

            this.profiler = new GrainProfiler(TaskScheduler.Current, providerRuntime);

            var dashboardGrain = providerRuntime.GrainFactory.GetGrain<IDashboardGrain>(0);
            return dashboardGrain.Init();
        }

        /// <summary> Init the factory.</summary>
        public virtual void Init(IProviderConfiguration config, string providerName, Logger logger)
        {
            if (config == null) throw new ArgumentNullException("config");
            if (!config.Properties.TryGetValue(DATA_CONNECTION_STRING, out dataConnectionString))
                throw new ArgumentException(String.Format("{0} property not set", DATA_CONNECTION_STRING));
            if (!config.Properties.TryGetValue(DEPLOYMENT_ID, out deploymentId))
                throw new ArgumentException(String.Format("{0} property not set", DEPLOYMENT_ID));
            
            string cacheSizeString;
            cacheSize = DEFAULT_CACHE_SIZE;
            if (config.Properties.TryGetValue(CACHE_SIZE_PARAM, out cacheSizeString))
            {
                if (!int.TryParse(cacheSizeString, out cacheSize))
                    throw new ArgumentException(String.Format("{0} invalid.  Must be int", CACHE_SIZE_PARAM));
            }
            string numQueuesString;
            numQueues = DEFAULT_NUM_QUEUES;
            if (config.Properties.TryGetValue(NUM_QUEUES_PARAM, out numQueuesString))
            {
                if (!int.TryParse(numQueuesString, out numQueues))
                    throw new ArgumentException(String.Format("{0} invalid.  Must be int", NUM_QUEUES_PARAM));
            }

            this.providerName = providerName;
            streamQueueMapper = new HashRingBasedStreamQueueMapper(numQueues, providerName);
            adapterCache = new SimpleQueueAdapterCache(this, cacheSize, logger);
        }

        /// <summary> Initialization function for this storage provider. </summary>
        /// <see cref="IProvider.Init"/>
        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Name = name;
            serviceId = providerRuntime.ServiceId.ToString();

            if (!config.Properties.ContainsKey(DATA_CONNECTION_STRING) || string.IsNullOrWhiteSpace(config.Properties[DATA_CONNECTION_STRING]))
                throw new ArgumentException("DataConnectionString property not set");

            dataConnectionString = config.Properties["DataConnectionString"];

            if (config.Properties.ContainsKey(TABLE_NAME_PROPERTY))
                tableName = config.Properties[TABLE_NAME_PROPERTY];

            isDeleteStateOnClear = config.Properties.ContainsKey(DELETE_ON_CLEAR_PROPERTY) &&
                "true".Equals(config.Properties[DELETE_ON_CLEAR_PROPERTY], StringComparison.OrdinalIgnoreCase);

            Log = providerRuntime.GetLogger("Storage.AzureTableStorage." + id);

            var initMsg = string.Format("Init: Name={0} ServiceId={1} Table={2} DeleteStateOnClear={3}",
                Name, serviceId, tableName, isDeleteStateOnClear);

            if (config.Properties.ContainsKey(USE_JSON_FORMAT_PROPERTY))
                useJsonFormat = "true".Equals(config.Properties[USE_JSON_FORMAT_PROPERTY], StringComparison.OrdinalIgnoreCase);
            
            if (useJsonFormat)
            {
                jsonSettings = jsonSettings = OrleansJsonSerializer.SerializerSettings;
            }
            initMsg = String.Format("{0} UseJsonFormat={1}", initMsg, useJsonFormat);

            Log.Info((int)AzureProviderErrorCode.AzureTableProvider_InitProvider, initMsg);
            Log.Info((int)AzureProviderErrorCode.AzureTableProvider_ParamConnectionString, "AzureTableStorage Provider is using DataConnectionString: {0}", ConfigUtilities.PrintDataConnectionInfo(dataConnectionString));
            tableDataManager = new GrainStateTableDataManager(tableName, dataConnectionString, Log);
            return tableDataManager.InitTableAsync();
        }

        public void Init(IProviderConfiguration config, string providerName, Logger logger, IServiceProvider serviceProvider)
        {
            if (config == null) throw new ArgumentNullException(nameof(config));
            if (logger == null) throw new ArgumentNullException(nameof(logger));
            if (String.IsNullOrEmpty(providerName)) throw new ArgumentNullException(nameof(providerName));

            // Creating an options object with all the config values
            _options = new KafkaStreamProviderOptions(config);

            if (!_options.UsingExternalMetrics)
            {
                Metric.Config.WithHttpEndpoint($"http://localhost:{_options.MetricsPort}/");
            }

            if (!_options.IncludeMetrics)
            {
                Metric.Context("KafkaStreamProvider").Advanced.CompletelyDisableMetrics();
            }

            _providerName = providerName;
            _streamQueueMapper = new HashRingBasedStreamQueueMapper(_options.NumOfQueues, providerName);
            _logger = logger;
            _adapter = new KafkaQueueAdapter(_streamQueueMapper, _options, providerName, new KafkaBatchFactory(), _logger);
            _adapterCache = new TimedQueueAdapterCache(this, TimeSpan.FromSeconds(_options.CacheTimespanInSeconds), _options.CacheSize, _options.CacheNumOfBuckets, logger);
        }

        /// <summary> Initialization function for this storage provider. </summary>
        /// <see cref="IProvider.Init"/>
        public async Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Log = providerRuntime.GetLogger("Storage.AzureBlobStorage");

            try
            {
                this.Name = name;
                ConfigureJsonSerializerSettings(config);

                if (!config.Properties.ContainsKey("DataConnectionString")) throw new BadProviderConfigException("The DataConnectionString setting has not been configured in the cloud role. Please add a DataConnectionString setting with a valid Azure Storage connection string.");

                var account = CloudStorageAccount.Parse(config.Properties["DataConnectionString"]);
                var blobClient = account.CreateCloudBlobClient();
                var containerName = config.Properties.ContainsKey("ContainerName") ? config.Properties["ContainerName"] : "grainstate";
                container = blobClient.GetContainerReference(containerName);
                await container.CreateIfNotExistsAsync().ConfigureAwait(false);

                Log.Info((int)AzureProviderErrorCode.AzureBlobProvider_InitProvider, "Init: Name={0} ServiceId={1} {2}", name, providerRuntime.ServiceId.ToString(), string.Join(" ", FormatPropertyMessage(config)));
                Log.Info((int)AzureProviderErrorCode.AzureBlobProvider_ParamConnectionString, "AzureBlobStorage Provider is using DataConnectionString: {0}", ConfigUtilities.PrintDataConnectionInfo(config.Properties["DataConnectionString"]));
            }
            catch (Exception ex)
            {
                Log.Error((int)AzureProviderErrorCode.AzureBlobProvider_InitProvider, ex.ToString(), ex);
                throw;
            }
        }

 public Task Init(Logger logger, IProviderConfiguration config, string providerName, int numQueues)
 {
     _logger = logger;
     _queues = new ConcurrentDictionary<QueueId, Queue<byte[]>>();
     IsInitialised = true;
     return TaskDone.Done;
 }

 public virtual void PopulateFromProviderConfig(IProviderConfiguration providerConfiguration)
 {
     EventHubSettingsType = providerConfiguration.GetTypeProperty(EventHubConfigTypeName, null);
     if (string.IsNullOrWhiteSpace(StreamProviderName))
     {
         throw new ArgumentOutOfRangeException("providerConfiguration", "StreamProviderName not set.");
     }
 }

        //---------------------------------------------------------------------
        public async Task Init(string name, IProviderRuntime provider_runtime, IProviderConfiguration config)
        {
            Console.Title = "GameCloud.IM v1.00.000 build20161014";
            Console.WriteLine("GameCloud.IM v1.00.000 build20161014");

            var startup = new IMStartup();
            await startup.Start();
        }

 public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
 {
     //process uncommit event message(store uncommit event message to event-store)
     //var eventDispatcher = GrainFactory.GetGrain<IEventStoreDispatcher>(0);
     GrainInternalEventHandlerProvider.RegisterInternalEventHandler(this.GetType().Assembly);
      
     return TaskDone.Done;
 }

 /// <summary>
 /// Initializes the provider during silo startup.
 /// </summary>
 /// <param name="name">The name of this provider instance.</param>
 /// <param name="providerRuntime">A Orleans runtime object managing all storage providers.</param>
 /// <param name="config">Configuration info for this provider instance.</param>
 /// <returns>Completion promise for this operation.</returns>
 public override Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
 {
     this.Name = name;
     this.RootDirectory = config.Properties["RootDirectory"];
     if (string.IsNullOrWhiteSpace(RootDirectory)) throw new ArgumentException("RootDirectory property not set");
     DataManager = new GrainStateFileDataManager(RootDirectory);
     return base.Init(name, providerRuntime, config);
 }

 // used only for testing
 internal async Task AddAndInitProvider(string name, IProvider provider, IProviderConfiguration config = null)
 {
     if (provider != null)
     {
         await provider.Init(name, this, config);
         statisticsProviderLoader.AddProvider(name, provider, config);
     }
 }

 /// <summary>
 /// Utility function to populate config from provider config
 /// </summary>
 /// <param name="providerConfiguration"></param>
 public virtual void PopulateFromProviderConfig(IProviderConfiguration providerConfiguration)
 {
     GeneratorConfigType = providerConfiguration.GetTypeProperty(GeneratorConfigTypeName, null);
     if (string.IsNullOrWhiteSpace(StreamProviderName))
     {
         throw new ArgumentOutOfRangeException("providerConfiguration", "StreamProviderName not set.");
     }
     TotalQueueCount = providerConfiguration.GetIntProperty(TotalQueueCountName, TotalQueueCountDefault);
 }

 /// <summary>
 /// Initializes the storage provider.
 /// </summary>
 /// <param name="name">The name of this provider instance.</param>
 /// <param name="providerRuntime">A Orleans runtime object managing all storage providers.</param>
 /// <param name="config">Configuration info for this provider instance.</param>
 /// <returns>Completion promise for this operation.</returns>
 public override Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
 {
     this.Name = name;
     this.ConnectionString = config.Properties["ConnectionString"];
     this.Database = config.Properties["Database"];
     if (string.IsNullOrWhiteSpace(ConnectionString)) throw new ArgumentException("ConnectionString property not set");
     if (string.IsNullOrWhiteSpace(Database)) throw new ArgumentException("Database property not set");
     DataManager = new GrainStateMongoDataManager(Database, ConnectionString);
     return base.Init(name, providerRuntime, config);
 }

 public void PopulateFromProviderConfig(IProviderConfiguration providerConfiguration)
 {
     EventHubSettingsType = providerConfiguration.GetTypeProperty(EventHubConfigTypeName, null);
     CheckpointSettingsType = providerConfiguration.GetTypeProperty(CheckpointSettingsTypeName, null);
     if (string.IsNullOrWhiteSpace(StreamProviderName))
     {
         throw new ArgumentOutOfRangeException("providerConfiguration", "StreamProviderName not set.");
     }
     CacheSizeMb = providerConfiguration.GetIntProperty(CacheSizeMbName, DefaultCacheSizeMb);
 }

        /// <summary> Init the factory.</summary>
        public virtual void Init(IProviderConfiguration config, string providerName, Logger logger, IServiceProvider serviceProvider)
        {
            if (config == null) throw new ArgumentNullException("config");
            if (!config.Properties.TryGetValue(AzureQueueAdapterFactory.DATA_CONNECTION_STRING, out dataConnectionString))
                throw new ArgumentException(String.Format("{0} property not set", AzureQueueAdapterFactory.DATA_CONNECTION_STRING));
            if (!config.Properties.TryGetValue(QUEUE_NAME_STRING, out queueName))
                throw new ArgumentException(String.Format("{0} property not set", QUEUE_NAME_STRING));

            this.providerName = providerName;
        }

 public async Task Init(Logger logger, IProviderConfiguration config, string providerName, int numQueues)
 {
     _logger = logger;
     _redisListBaseName = $"orleans-{providerName}-queue";
     ReadRedisConnectionParams(config);
     _connection = await ConnectionMultiplexer.ConnectAsync(_server);
     _database = _connection.GetDatabase(_databaseNum);
     logger.AutoInfo($"connection to Redis successful.");
     IsInitialised = true;
 }