/// <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(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;
        }

        /// <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 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)
        {
            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();
        }

        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));
            }
        }

        //---------------------------------------------------------------------
        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);
 }

        public async Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Name = name;
            logger = providerRuntime.GetLogger("SqlStatisticsPublisher");

            //TODO: Orleans does not yet provide the type of database used (to, e.g., to load dlls), so SQL Server is assumed.            
            database = RelationalStorageUtilities.CreateGenericStorageInstance(WellKnownRelationalInvariants.SqlServer, config.Properties["ConnectionString"]);

            await InitOrleansQueriesAsync();           
        }

 /// <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 Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            providerRuntime.SetInvokeInterceptor((method, request, grain, invoker) =>
            {
                RequestContext.Set("GrainInfo", 38);
                return invoker.Invoke(grain, request);
            });

            return Task.FromResult(0);
        }

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

            string latencyParam = config.Properties[LATENCY_PARAM_STRING];
            latency = latencyParam == null ? DefaultLatency : TimeSpan.Parse(latencyParam);
            Log.Info("Init: Fixed Store Latency={0}", latency);

            mockCallsOnly = config.Properties.ContainsKey(MOCK_CALLS_PARAM_STRING) &&
                "true".Equals(config.Properties[MOCK_CALLS_PARAM_STRING], StringComparison.OrdinalIgnoreCase);
        }

        /// <summary>
        /// Initializes publisher
        /// </summary>
        /// <param name="name">Provider name</param>
        /// <param name="providerRuntime">Provider runtime API</param>
        /// <param name="config">Provider configuration</param>
        /// <returns></returns>
        public async Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Name = name;
            logger = providerRuntime.GetLogger("SqlStatisticsPublisher");

            string adoInvariant = AdoNetInvariants.InvariantNameSqlServer;
            if (config.Properties.ContainsKey("AdoInvariant"))
                adoInvariant = config.Properties["AdoInvariant"];

            orleansQueries = await RelationalOrleansQueries.CreateInstance(adoInvariant, config.Properties["ConnectionString"]);
        }

        public GrainController(Router router, TaskScheduler taskScheduler, IProviderRuntime providerRuntime)
        {
            this.TaskScheduler = taskScheduler;
            this.ProviderRuntime = providerRuntime;

            Action<string, Func<IOwinContext, IDictionary<string, string>, Task>> add = router.Add;

            add("/grain/:type/:id/:method", CallGrain);
            add("/grain/:type/:id/:method/:classprefix", CallGrain);
            add("/pinggrain", PingGrain);
            add("/ping", Ping);
        }

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

            Console.WriteLine("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            Console.WriteLine("IMTestAppOrleansBootstrap.Init()");
            Console.WriteLine("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
            IMTestAppContext.Instance.Setup();

            var grain_player = provider_runtime.GrainFactory.GetGrain<IGrainIMTestAppPlayer>(Guid.NewGuid());
            await grain_player.Setup();

        }

        public async Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Name = name;
            logger = providerRuntime.GetLogger("SqlStatisticsPublisher");

            string adoInvariant = AdoNetInvariants.InvariantNameSqlServer;
            if (config.Properties.ContainsKey("AdoInvariant"))
                adoInvariant = config.Properties["AdoInvariant"];

            database = RelationalStorageUtilities.CreateGenericStorageInstance(adoInvariant, config.Properties["ConnectionString"]);

            queryConstants = await database.InitializeOrleansQueriesAsync(); 
        }

        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            var system = ClusterActorSystem.Current;
            var providers = config.Properties["providers"].Split(';');

            StreamPubSubWrapper.Hook(providers, stream => 
                StreamSubscriptionMatcher
                    .Match(system, stream)
                    .Select(x => new StreamPubSubMatch(x.Receive))
                    .ToArray());

            return TaskDone.Done;
        }

        public GrainProfiler(TaskScheduler taskScheduler, IProviderRuntime providerRuntime)
        {
            this.TaskScheduler = taskScheduler;
            this.ProviderRuntime = providerRuntime;

            // register interceptor, wrapping any previously set interceptor
            this.innerInterceptor = providerRuntime.GetInvokeInterceptor();
            providerRuntime.SetInvokeInterceptor(this.InvokeInterceptor);
            siloAddress = providerRuntime.SiloIdentity.ToSiloAddress();

            // register timer to report every second
            timer = new Timer(this.ProcessStats, providerRuntime, 1 * 1000, 1 * 1000);

        }

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

            ConnectString = config.Properties["DataConnectionString"];

            if (config.Properties.ContainsKey("Table"))
            {
                Table = config.Properties["Table"];
                CustomTable = true;
            }

            return TaskDone.Done;
        }