public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            var batch = new TableBatchOperation();

            foreach(var msg in messages)
            {
                var snap = JsonConvert.DeserializeObject<BusSnapshotInfo>(Encoding.UTF8.GetString(msg.GetBytes()));

                var entity = new DynamicTableEntity(snap.RouteShortName, snap.VehicleId.ToString());

                entity.Properties.Add("RouteShortName", EntityProperty.GeneratePropertyForString(snap.RouteShortName));
                entity.Properties.Add("VehicleId", EntityProperty.GeneratePropertyForInt(snap.VehicleId));
                entity.Properties.Add("TripId", EntityProperty.GeneratePropertyForInt(snap.TripId));
                entity.Properties.Add("Latitude", EntityProperty.GeneratePropertyForDouble(snap.Latitude));
                entity.Properties.Add("Longitude", EntityProperty.GeneratePropertyForDouble(snap.Longitude));
                entity.Properties.Add("DirectionOfTravel", EntityProperty.GeneratePropertyForString(snap.DirectionOfTravel.ToString()));
                entity.Properties.Add("NextStopId", EntityProperty.GeneratePropertyForInt(snap.NextStopId));
                entity.Properties.Add("Timeliness", EntityProperty.GeneratePropertyForString(snap.Timeliness.ToString()));
                entity.Properties.Add("TimelinessOffset", EntityProperty.GeneratePropertyForInt(snap.TimelinessOffset));
                entity.Properties.Add("Timestamp", EntityProperty.GeneratePropertyForDateTimeOffset(snap.Timestamp));

                batch.Add(TableOperation.InsertOrReplace(entity));
            }

            var tableClient = _account.CreateCloudTableClient();

            var table = tableClient.GetTableReference("snapshots");

            await table.CreateIfNotExistsAsync();

            await table.ExecuteBatchAsync(batch);

            await context.CheckpointAsync();
        }

 //------------------------------------------------------------------------------------------------------------------------
 Task IEventProcessor.OpenAsync(PartitionContext context)
 {
     DebugEx.TraceLog("SimpleEventProcessor initialized.  Partition: " + context.Lease.PartitionId + ", Offset: " + context.Lease.Offset);
     this.checkpointStopWatch = new Stopwatch();
     this.checkpointStopWatch.Start();
     return Task.FromResult<object>(null);
 }

        async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {

            foreach (EventData eventData in messages)
            {
                _Logger.LogInfo(string.Format("Event received from partition: {0} - {1}", context.Lease.PartitionId,eventData.PartitionKey));

                try
                {
                    var httpMessage = HttpMessage.Parse(eventData.GetBodyStream());
                    await _MessageContentProcessor.ProcessHttpMessage(httpMessage);
                }
                catch (Exception ex)
                {
                    _Logger.LogError(ex.Message);
                }
            }

            //Call checkpoint every 5 minutes, so that worker can resume processing from the 5 minutes back if it restarts.
            if (this.checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
            {
                _Logger.LogInfo("Checkpointing");
               await context.CheckpointAsync();
                this.checkpointStopWatch.Restart();
            }
        }

 public async Task OpenAsync(PartitionContext context)
 {
     if (!WebJobsHelper.RunAsWebJobs)
         Console.WriteLine(string.Format("EventProcessor initialization. Partition: '{0}', Offset: '{1}'",
             context.Lease.PartitionId, context.Lease.Offset));
     partitionContext = context;
     var retries = 3;
     while (retries > 0)
     {
         try
         {
             retries--;
             hubClient = EventHubClient.CreateFromConnectionString(
                 ConfigurationManager.ConnectionStrings["SigfoxDemoAlertSender"].ConnectionString,
                 "alert");
             cacheConnection = await ConnectionMultiplexer.ConnectAsync(ConfigurationManager.ConnectionStrings["SigfoxDemoCache"].ConnectionString);
             cacheDatabase = cacheConnection.GetDatabase();
             sqlConnection = new SqlConnection(ConfigurationManager.ConnectionStrings["SigfoxDemoDatabase"].ConnectionString);
             //sqlConnection.Open();
             //sqlCommand = new SqlCommand("InsertAlert", sqlConnection) { CommandType = CommandType.StoredProcedure };
             //sqlCommand.Parameters.Add(new SqlParameter("@Device", SqlDbType.VarChar));
             retries = 0;
         }
         catch (Exception e)
         {
             Console.Error.WriteLine("Error opening destination Event Hub: " + e.Message);
             if (retries == 0)
                 throw;
         }
     }
     checkpointStopWatch = new Stopwatch();
     checkpointStopWatch.Start();
 }

		public async Task CloseAsync(PartitionContext context, CloseReason reason)
		{
			_log.InfoEvent("Close",
				new Facet("reason", reason),
				new Facet("partitionId", context.Lease.PartitionId),
				new Facet("offset", context.Lease.Offset));
		}

		public async Task OpenAsync(PartitionContext context)
		{
			_log.InfoEvent("Open",
				new Facet("eventHubPath", context.EventHubPath),
				new Facet("partitionId", context.Lease.PartitionId),
				new Facet("offset", context.Lease.Offset));
		}

        public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
        {
            // Workaround for event hub sending null on timeout
            events = events ?? Enumerable.Empty<EventData>();

            if(!await _elasticSearchWriter.WriteAsync(events.ToList(), _token).ConfigureAwait(false))
            {
                return;
            }

            try
            {
                EventData checkpointEventData = events.LastOrDefault();

                await context.CheckpointAsync(checkpointEventData);

                WarmStorageEventSource.Log.CheckpointCompleted(ProcessorName, _eventHubName, context.Lease.PartitionId, checkpointEventData.Offset);
            }
            catch (Exception ex)
            {
                if (!(ex is StorageException || ex is LeaseLostException))
                {
                    throw;
                }

                WarmStorageEventSource.Log.UnableToCheckpoint(ex, ProcessorName, _eventHubName, context.Lease.PartitionId);
            }
        }

        public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            Trace.TraceInformation("\n");
            Trace.TraceInformation("........ProcessEventsAsync........");
            foreach (EventData eventData in messages)
            {
                try
                {
                    string jsonString = Encoding.UTF8.GetString(eventData.GetBytes());

                    Trace.TraceInformation(string.Format("Message received at '{0}'. Partition: '{1}'",
                        eventData.EnqueuedTimeUtc.ToLocalTime(), this.partitionContext.Lease.PartitionId));

                    Trace.TraceInformation(string.Format("-->Raw Data: '{0}'", jsonString));

                    SensorEvent newSensorEvent = this.DeserializeEventData(jsonString);

                    Trace.TraceInformation(string.Format("-->Serialized Data: '{0}', '{1}', '{2}', '{3}', '{4}'",
                        newSensorEvent.timestart, newSensorEvent.dsplalert, newSensorEvent.alerttype, newSensorEvent.message, newSensorEvent.targetalarmdevice));

                    // Issuing alarm to device.
                    string commandParameterNew = "{\"Name\":\"AlarmThreshold\",\"Parameters\":{\"SensorId\":\"" + newSensorEvent.dsplalert + "\"}}";
                    Trace.TraceInformation("Issuing alarm to device: '{0}', from sensor: '{1}'", newSensorEvent.targetalarmdevice, newSensorEvent.dsplalert);
                    Trace.TraceInformation("New Command Parameter: '{0}'", commandParameterNew);
                    await WorkerRole.iotHubServiceClient.SendAsync(newSensorEvent.targetalarmdevice, new Microsoft.Azure.Devices.Message(Encoding.UTF8.GetBytes(commandParameterNew)));
                }
                catch (Exception ex)
                {
                    Trace.TraceInformation("Error in ProssEventsAsync -- {0}\n", ex.Message);
                }
            }

            await context.CheckpointAsync();
        }

 Task IEventProcessor.OpenAsync(PartitionContext context)
 {
     Console.WriteLine("SimpleEventProcessor initialized.  Partition: '{0}', Offset: '{1}'", context.Lease.PartitionId, context.Lease.Offset);
     this.checkpointStopWatch = new Stopwatch();
     this.checkpointStopWatch.Start();
     return Task.FromResult<object>(null);
 }

 public Task OpenAsync(PartitionContext context)
 {
     Console.WriteLine("EventProcessor started");
     this.stopWatch = new Stopwatch();
     this.stopWatch.Start();
     return Task.FromResult<object>(null);
 }

        async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            var iDbService = new DbService();
            _client = iDbService.GetFirebaseClient();
            foreach (EventData eventData in messages)
            {
                string data = Encoding.UTF8.GetString(eventData.GetBytes());
                FirebaseResponse response = await _client.PushAsync("event", new EHdata
                {
                    offset = eventData.Offset,
                    body = data,
                    partitionId = context.Lease.PartitionId
                });
                Console.WriteLine(String.Format("Message received.  Partition: '{0}', Data: '{1}', Offset: '{2}'",
                    context.Lease.PartitionId, data, eventData.Offset));
            }

            //Call checkpoint every 5 minutes, so that worker can resume processing from the 5 minutes back if it restarts.
            if (this.checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
            {
                Console.WriteLine(this.checkpointStopWatch.Elapsed);
                await context.CheckpointAsync();
                this.checkpointStopWatch.Restart();
            }
        }

 public Task OpenAsync(PartitionContext context)
 {
     this.partitionContext = context;
     this.checkpointStopWatch = new Stopwatch();
     this.checkpointStopWatch.Start();
     return Task.FromResult<object>(null);
 }

 async Task IEventProcessor.CloseAsync(PartitionContext context, CloseReason reason)
 {
     if (reason == CloseReason.Shutdown)
     {
         await context.CheckpointAsync();
     }
 }

 public IEventProcessor CreateEventProcessor(PartitionContext context)
 {
     var processor = new DeviceAdministrationProcessor(_deviceLogic, _configurationProvider);
     processor.ProcessorClosed += this.ProcessorOnProcessorClosed;
     this.eventProcessors.TryAdd(context.Lease.PartitionId, processor);
     return processor;
 }

        async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            foreach (EventData eventData in messages)
            {
                byte[] data = eventData.GetBytes();

                if (eventData.Properties.ContainsKey("messageType") && (string)eventData.Properties["messageType"] == "interactive")
                {
                    var messageId = (string)eventData.SystemProperties["message-id"];

                    var queueMessage = new BrokeredMessage(new MemoryStream(data));
                    queueMessage.MessageId = messageId;
                    queueMessage.Properties["messageType"] = "interactive";
                    await queueClient.SendAsync(queueMessage);

                    WriteHighlightedMessage(string.Format("Received interactive message: {0}", messageId));
                    continue;
                }

                if (toAppend.Length + data.Length > MAX_BLOCK_SIZE || stopwatch.Elapsed > MAX_CHECKPOINT_TIME)
                {
                    await AppendAndCheckpoint(context);
                }
                await toAppend.WriteAsync(data, 0, data.Length);

                Console.WriteLine(string.Format("Message received.  Partition: '{0}', Data: '{1}'",
                  context.Lease.PartitionId, Encoding.UTF8.GetString(data)));
            }
        }

        public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
        {
            // Workaround for event hub sending null on timeout
            events = events ?? Enumerable.Empty<EventData>();

            foreach (var eventData in events)
            {
                var updateTemperatureEvent = JsonConvert.DeserializeObject<UpdateTemperatureEvent>(Encoding.UTF8.GetString(eventData.GetBytes()));
                eventData.Properties["BuildingId"] = _buildingLookupService.GetBuildingId(updateTemperatureEvent.DeviceId);
            }

            if(!await _elasticSearchWriter.WriteAsync(events.ToList(), _token).ConfigureAwait(false))
            {
                return;
            }

            try
            {
                EventData checkpointEventData = events.LastOrDefault();

                await context.CheckpointAsync(checkpointEventData);

                WarmStorageEventSource.Log.CheckpointCompleted(ProcessorName, _eventHubName, context.Lease.PartitionId, checkpointEventData.Offset);
            }
            catch (Exception ex)
            {
                if (!(ex is StorageException || ex is LeaseLostException))
                {
                    throw;
                }

                WarmStorageEventSource.Log.UnableToCheckpoint(ex, ProcessorName, _eventHubName, context.Lease.PartitionId);
            }
        }

        async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            var partitionedMessages = messages.GroupBy(data => data.PartitionKey).ToDictionary(datas => datas.Key, datas => datas.ToList());

            //For each partition spawn a Task which will sequentially iterate over its own block
            //Wait for all Tasks to complete, before proceeding.
            await Task.WhenAll(partitionedMessages.Select(partition => Task.Run(async () =>
            {
                var block = partition.Value;
                foreach (var eventData in block)
                {
                    try
                    {
                        var data = Encoding.UTF8.GetString(eventData.GetBytes());

                        System.Console.WriteLine(DateTime.Now + ":Message received.  Partition: '{0}', Data: '{1}', Partition Key: '{2}'", context.Lease.PartitionId, data, eventData.PartitionKey);
                    }
                    catch (Exception e)
                    {
                        //do something with your logs..
                    }
                }
            })));

            //Call checkpoint every 5 minutes, so that worker can resume processing from the 5 minutes back if it restarts.
            if (checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
            {
                await context.CheckpointAsync();
                checkpointStopWatch.Restart();
            }
        }

        public Task OpenAsync(PartitionContext context)
        {
            /*client = new HttpClient();
            client.DefaultRequestHeaders.Add("X-ZUMO-APPLICATION", APP_KEY_MOBILE_SERVICES);
            client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
            return Task.FromResult<object>(null);*/

            var storageAccount = CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
            var tableClient = storageAccount.CreateCloudTableClient();
            sensorLogTable = tableClient.GetTableReference("SensorLog");
            //sensorLogTable = tableClient.GetTableReference("SensorLog" + context.Lease.PartitionId);

            // 閾値の取得
            if (thresholdTempWarning == null)
            {
                var sensorConfigTable = tableClient.GetTableReference("SensorConfig");
                var query = new TableQuery<SensorConfig>();
                var configData = sensorConfigTable.ExecuteQuery(query);
                foreach (SensorConfig config in configData)
                {
                    if (config.PartitionKey == "TemperatureWarning")
                    {
                        thresholdTempWarning = config.Threshold;
                        System.Console.WriteLine("ThresholdTempWarning: " + thresholdTempWarning);
                    }
                    else if (config.PartitionKey == "TemperatureDanger")
                    {
                        thresholdTempDanger = config.Threshold;
                        System.Console.WriteLine("ThresholdTempDanger: " + thresholdTempDanger);
                    }
                }
            }

            return sensorLogTable.CreateIfNotExistsAsync();
        }

        public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            foreach (EventData eventData in messages)
            {
                if (eventData.Properties.ContainsKey("time"))
                {
                    if (eventData.Properties.ContainsKey("temp"))
                        Console.WriteLine(string.Format("time = {0}, temp = {1}", eventData.Properties["time"], eventData.Properties["temp"]));

                    if (eventData.Properties.ContainsKey("hmdt"))
                        Console.WriteLine(string.Format("time = {0}, hmdt = {1}", eventData.Properties["time"], eventData.Properties["hmdt"]));

                    if (eventData.Properties.ContainsKey("accx") &&
                        eventData.Properties.ContainsKey("accy") &&
                        eventData.Properties.ContainsKey("accz"))
                        Console.WriteLine(string.Format("time = {0}, accx = {1}, accy = {2}, accz = {3}", eventData.Properties["time"], eventData.Properties["accx"], eventData.Properties["accy"], eventData.Properties["accz"]));

                    if (eventData.Properties.ContainsKey("bpm"))
                        Console.WriteLine(string.Format("time = {0}, bpm = {1}", eventData.Properties["time"], eventData.Properties["bpm"]));
                }
            }

            //Call checkpoint every 5 minutes, so that worker can resume processing from the 5 minutes back if it restarts.
            //if (this.checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
            if (this.checkpointStopWatch.Elapsed > TimeSpan.FromSeconds(30))
            {
                await context.CheckpointAsync();
                this.checkpointStopWatch.Restart();
            }
        }

 public async Task CloseAsync(PartitionContext context, CloseReason reason)
 {
     if (reason == CloseReason.Shutdown)
     {
         await context.CheckpointAsync();
     }
 }