I have done the same process using generic class. Using function app i have achieved this. please find the following codes.
First i have created a function app MyEventhubTrigger.cs
using Microsoft.ApplicationInsights;
using Microsoft.ApplicationInsights.DataContracts;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.ServiceBus;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System;
using AzureMonitorFunctions.WebHook;
namespace MyFunctions
{
public static class MyEventhubTrigger
{
[FunctionName("MyEventhubTrigger")]
public static async void Run([EventHubTrigger("%EventHubName%", Connection = "EventHubConnectionString", ConsumerGroup = "devicestatuscg")] string myEventHubMessage, DateTime enqueuedTimeUtc, Int64 sequenceNumber, string offset, ILogger log)
{
log.LogInformation($"C# Event Hub trigger function processed a message: {myEventHubMessage}");
try
{
StorageTableOperation<StorageTable> tableObj = new StorageTableOperation<StorageTable>("EventhubMessages");
Eventhubstorage Eventhubdada = tableObj.GetTableData("Eventhub", "DeviceStatus");
if (Eventhubdada.EnqueuedTime < enqueuedTimeUtc)
{
Eventhubdada.EnqueuedTime = enqueuedTimeUtc;
Eventhubdada.Offset = offset;
Eventhubdada.Sequence = sequenceNumber;
await tableObj.UpdateTableData("Eventhub", "DeviceStatus", Eventhubdada);
}
}
}
catch (Exception ex)
{
log.LogInformation(ex.ToString());
}
}
}
}
Now i have create table storage in azure storage account. And add the same field in a property file. which is StorageTable.cs
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.WindowsAzure.Storage.Table;
namespace MyFunctions
{
public class StorageTable : TableEntity
{
public StorageTable()
{
}
public StorageTable(string PKey, string RKey)
{
PartitionKey = PKey;
RowKey = RKey;
}
public DateTime EnqueuedTime { get; set; }
public Int64 Sequence { get; set; }
public string Offset { get; set; }
}
}
Now I have created a class which can take any TableEntity property class and do the following operation Get/insert/Update. Find the following code for StorageTableOperation.cs
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Table;
using System.Threading.Tasks;
namespace MyFunctions
{
public class StorageTableOperation<T> where T : ITableEntity
{
internal CloudTable cloudTable;
public StorageTableOperation(string tableName)
{
string connectionString = Environment.GetEnvironmentVariable("AzureWebJobsStorage");
CloudStorageAccount storageAccount = CloudStorageAccount.Parse(connectionString);
CloudTableClient client = storageAccount.CreateCloudTableClient();
cloudTable = client.GetTableReference(tableName);
}
public T GetTableData(string PartitionKey,string RowKey)
{
TableOperation retOp = TableOperation.Retrieve<T>(PartitionKey, RowKey);
TableResult result = cloudTable.Execute(retOp);
T tableData = (T)result.Result;
return tableData;
}
public async Task<bool> UpdateTableData(string PartitionKey, string RowKey, T tableData)
{
try
{
TableOperation operation = TableOperation.InsertOrMerge(tableData);
TableResult tableResult = await cloudTable.ExecuteAsync(operation);
return true;
}
catch(Exception ex)
{
return false;
}
}
}
}