gistfile1.cs

void CreateClients(MigrationMessage migrationMessage)
        {
            var agencyUniqueId = migrationMessage.AgencyUniqueId;
            var agency = agencyService.GetAgencyByUniqueId(agencyUniqueId);

            if (agency == null)
                throw new ApplicationException("Agency unique id provided for the import process does not match any existing agencies.");

            using (var connection = new SqlCeConnection(connectionString))
            using (var canonicalDb = new ImportDbContext(connection))
            {
                var clients = canonicalDb.Clients;
                var count = 0;
                var total = clients.Count();
                Log("Importing {0} clients into the database", total);
                var startedAt = DateTimeOffset.UtcNow;
                var lastLoggedAt = DateTimeOffset.UtcNow;
                var statusLogLock = new object();

                circuitBreaker.StateChanged += CircuitBreakerStateChanged;
                circuitBreaker.ServiceLevelChanged += CircuitBreakerServiceLevelChanged;

                const ushort allowedRetries = 20;
                var retryInterval = TimeSpan.FromSeconds(30);
                var processorCount = Environment.ProcessorCount;
                var threadsToAllocate = (int)Math.Ceiling(processorCount*10 * 0.9);

                var options = new ParallelOptions { MaxDegreeOfParallelism = threadsToAllocate };
                Parallel.ForEach(clients, options, client =>
                {
                    Interlocked.Increment(ref count);

                    var startTime = DateTime.UtcNow;
                    try
                    {
                        circuitBreaker.ExecuteWithRetries(() => TryImportClient(agency, client, startTime),
                            allowedRetries,
                            retryInterval);
                    }
                    catch (Exception ex)
                    {
                        Log("Failed importing client\r\n" +
                            "Client object was: {0}\r\n" +
                            "Exception was: {1}",
                            SerializationHelper.Serialize(client),
                            ex.ToString());
                        throw;
                    }

                    LogStatusEveryThousandClientsOrEveryThirtySeconds(ref count, total, statusLogLock, startedAt, ref lastLoggedAt);
                });

                Log("All {0} clients imported", total);
            }
        }