EventHubsTest.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.EventHubs;
using Newtonsoft.Json;

namespace ConsoleApp2
{
    public class Program
    {
        private static EventHubClient _eventHubClient;

        private const string EventHubConnectionString = "";
        private const string EventHubName = "wifis";

        private static int _counter = 0;

        private static readonly JsonSerializerSettings JsonSerializerSettings = new JsonSerializerSettings()
        {
            NullValueHandling = NullValueHandling.Ignore
        };

        private static readonly Random Random = new Random();

        static Program()
        {
            var connectionStringBuilder = new EventHubsConnectionStringBuilder(EventHubConnectionString)
            {
                EntityPath = EventHubName
            };

            _eventHubClient = EventHubClient.CreateFromConnectionString(connectionStringBuilder.ToString());
        }

        public static async Task Main2(string[] args)
        {
            while (true)
            {
                var startNew = Stopwatch.StartNew();
                _counter = 0;

                await SendRequestsAsync(requestsCount: 150, minMessagesPerRequest: 50);

                var milliseconds = Math.Max(startNew.ElapsedMilliseconds, 1);

                decimal messagesPerSecond = _counter / (decimal)milliseconds * 1000;
                Console.WriteLine($"\r\n\r\nThroughput: {messagesPerSecond:.#} m/sec\r\n\r\n");
            }
        }

        private static async Task SendRequestsAsync(int requestsCount, int minMessagesPerRequest)
        {
            var requestsTasks = Enumerable.Range(0, requestsCount).Select(index =>
            {
                int messagesPerThisRequest = Random.Next(minMessagesPerRequest, minMessagesPerRequest + 100 + index);

                return SendRequestUsingBatchAsync(messagesPerThisRequest, index);
                // return SendRequestUsingParallelAsync(messagesPerThisRequest, index);
            }).ToArray();

            await Task.WhenAll(requestsTasks).ConfigureAwait(false);
        }

        private static async Task SendRequestUsingParallelAsync(int numMessagesToSend, int requestId)
        {
            var eventHubClient = GetClient();

            var requests = Enumerable.Range(0, numMessagesToSend).Select(_ => SendOneMessageAsync(eventHubClient, requestId)).ToArray();

            try
            {
                await Task.WhenAll(requests).ConfigureAwait(false);
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }
            finally
            {
                await eventHubClient.CloseAsync();
            }
        }

        private static async Task SendRequestUsingBatchAsync(int numMessagesToSend, int requestId)
        {
            var events = Enumerable.Range(0, numMessagesToSend)
                .Select(_ => WifiLocationAnalytic.CreateRandom())
                .Select(ToEventData)
                .ToArray();

            var eventHubClient = GetClient();
            try
            {
                await eventHubClient.SendAsync(events).ConfigureAwait(false);
            }
            catch (Exception e)
            {
                Console.WriteLine(e);

            }
            finally
            {
                await eventHubClient.CloseAsync();
            }

            Interlocked.Add(ref _counter, events.Length);
        }

        private static EventHubClient GetClient()
        {
            // return _eventHubClient;
            var connectionStringBuilder = new EventHubsConnectionStringBuilder(EventHubConnectionString)
            {
                EntityPath = EventHubName
            };

            var eventHubClient = EventHubClient.CreateFromConnectionString(connectionStringBuilder.ToString());
            return eventHubClient;
        }

        private static Task SendOneMessageAsync(EventHubClient eventHubClient, int requestId)
        {
            return SendOneMessageAsync(eventHubClient, requestId, WifiLocationAnalytic.CreateRandom());
        }

        private static async Task SendOneMessageAsync(EventHubClient eventHubClient, int requestId, WifiLocationAnalytic message)
        {
            var pKey = message.Id;

            await eventHubClient.SendAsync(ToEventData(message), pKey);

            Interlocked.Increment(ref _counter);
        }

        private static EventData ToEventData(WifiLocationAnalytic message)
        {
            var json = JsonConvert.SerializeObject(message, Formatting.None, JsonSerializerSettings);

            return new EventData(Encoding.UTF8.GetBytes(json));
        }

        public class WifiLocationAnalytic
        {
            public string Id { get; set; }
            public string AccessPointMacAddress { get; set; }
            public List<string> AccessPointFloors { get; set; }
            public List<string> AccessPointTags { get; set; }
            public string ClientMacAddress { get; set; }
            public string IpAddressV4 { get; set; }
            public string IpAddressV6 { get; set; }
            public DateTime? SeenTime { get; set; }
            public long? SeenEpoch { get; set; }
            public string Ssid { get; set; }
            public int? Rssi { get; set; }
            public string Manufacturer { get; set; }
            public string OperatingSystem { get; set; }
            public decimal? Latitude { get; set; }
            public decimal? Longitude { get; set; }
            public decimal? Uncertainty { get; set; }
            public List<decimal> XOffsets { get; set; }
            public List<decimal> YOffsets { get; set; }
            public string TimeStamp { get; set; }
            public System.DateTime CreateDate { get; set; }
            public DateTime? EditDate { get; set; }
            public DateTime? DeleteDate { get; set; }

            private static readonly Random Random = new Random();
            private static readonly string[] _deviceMacs;
            private static readonly string[] _accessPointMacs;
            private static readonly string[] _ips;
            private static readonly PointF[] _locations;
            private static readonly string[] _oses;
            private static readonly string[] _floors;

            public static WifiLocationAnalytic CreateRandom()
            {
                var pointF = GetRandom(_locations);
                return new WifiLocationAnalytic
                {
                    Id = Guid.NewGuid().ToString(),
                    ClientMacAddress = GetRandom(_deviceMacs),
                    AccessPointMacAddress = GetRandom(_accessPointMacs),
                    IpAddressV4 = GetRandom(_ips),
                    Latitude = (decimal?)pointF.X,
                    Longitude = (decimal?)pointF.Y,
                    SeenEpoch = DateTimeOffset.UtcNow.ToUnixTimeSeconds(),
                    OperatingSystem = GetRandom(_oses),
                    AccessPointFloors = new List<string>(1) { GetRandom(_floors) },
                    CreateDate = DateTime.UtcNow,
                    TimeStamp = DateTime.UtcNow.ToLongDateString(),
                    SeenTime = DateTime.UtcNow
                };
            }

            static WifiLocationAnalytic()
            {
                _deviceMacs = GetMacs(50000);
                _accessPointMacs = GetMacs(300);
                _ips = GetIps(1000);
                _locations = GetLocations(100000);
                _oses = GetOses();
                _floors = GetFloors();
            }

            private static T GetRandom<T>(T[] array)
            {
                return array[Random.Next(0, array.Length)];
            }

            private static PointF[] GetLocations(int count)
            {
                var hashSet = new HashSet<PointF>();

                while (hashSet.Count < count)
                {
                    var lat = Random.Next(70, 80);
                    var lon = Random.Next(50, 60);

                    hashSet.Add(new PointF((float)(lat + Random.NextDouble()), (float)(lon + Random.NextDouble())));
                }

                return hashSet.ToArray();
            }

            private static string[] GetOses()
            {
                return new[]
                {
                    "Windows Server 2012", "Windows Server 2016", "Windows 8.1",
                    "Windows 10", "MacOS", "Android 5.0", "Android 7.0", "Android 8.0", "IOS 6", "IOS 7", "IOS 8",
                    "IOS 9", "IOS 10"
                };
            }

            private static string[] GetFloors()
            {
                return new[]
                {
                    "1st floor",
                    "2nd floor",
                    "3d floor",
                    "4th floor",
                    "roof top"
                };
            }

            private static string[] GetMacs(int count)
            {
                var hashSet = new HashSet<string>(StringComparer.Ordinal);

                while (hashSet.Count < count)
                {
                    hashSet.Add(GetRandomMac());
                }

                return hashSet.ToArray();
            }

            private static string[] GetIps(int count)
            {
                var hashSet = new HashSet<string>(StringComparer.Ordinal);

                while (hashSet.Count < count)
                {
                    hashSet.Add(GetRandomIp());
                }

                return hashSet.ToArray();
            }

            private static string GetRandomIp()
            {
                var v1 = Random.Next(0, 256);
                var v2 = Random.Next(0, 256);
                var v3 = Random.Next(0, 256);
                var v4 = Random.Next(0, 256);

                return $"{v1}.{v2}.{v3}.{v4}";
            }

            private static string GetRandomMac()
            {
                string guid = Guid.NewGuid().ToString("N");

                var result = new char[6 * 2 + 5];

                for (int i = 0; i < result.Length; i++)
                {
                    if ((i + 1) % 3 == 0)
                    {
                        result[i] = ':';
                    }
                    else
                    {
                        result[i] = guid[i];
                    }
                }

                return new string(result);
            }
        }
    }
}