GetAsyncWithCancellation2

GetAsyncWithCancellation2.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using Lextm.SharpSnmpLib;
using Lextm.SharpSnmpLib.Messaging;
using Lextm.SharpSnmpLib.Security;

namespace SandboxForGetAsync
{
    internal static class Program
    {
        private static readonly NumberGenerator RequestCounter = new NumberGenerator(0, 5000);
        private static readonly IPEndPoint RemoteEp = new IPEndPoint(IPAddress.Parse("10.161.227.164"), 161);
        private static readonly OctetString Community = new OctetString("public");
        private static readonly ObjectIdentifier Oid = new ObjectIdentifier("1.3.6.1.2.1.1.1.0");
        private const int TestCount = 100;

        internal static int Main(string[] args)
        {
            try
            {
                return Program.InternalMainAsync(args).Result;
            }
            catch (Exception ex)
            {
                Console.Error.WriteLine(ex);
                return Marshal.GetHRForException(ex);
            }
            finally
            {
                Console.WriteLine("hit any key to exit...");
                // shapshot(#3) here on dotMemory GUI.
                Console.ReadLine();
            }
        }

        private static async Task<int> InternalMainAsync(string[] args)
        {
            // await Program.PrepareTestMemory1();
            await Program.PrepareTestMemory2();

            Console.WriteLine("hit any key to start test 1...");
            // shapshot(#1) here on dotMemory GUI.
            Console.ReadLine();

            // await Program.TestMemory1();
            await Program.TestMemory2();

            Console.WriteLine("hit any key to start test 2...");
            // shapshot(#2) here on dotMemory GUI.
            Console.ReadLine();

            // await Program.TestMemory1();
            await Program.TestMemory2();

            return 0;
        }

        #region Test1 (SnmpClient sample)

        private static async Task PrepareTestMemory1()
        {
            await Program.TestToCallSnmpClient1(-1);
        }

        private static async Task TestMemory1()
        {
            // Running this code on Host-A
            // The Snmp Agent on Host-B, but an agent is not listening...

            var tasks = Enumerable.Range(0, Program.TestCount).Select(async (_, index) => await Program.TestToCallSnmpClient1(index));
            var allTasks = Task.WhenAll(tasks);
            try
            {
                await allTasks.ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Console.WriteLine("{0}", ex.InnerException);
            }
        }

        private static async Task TestToCallSnmpClient1(int index)
        {
            // request: RemoteEP=Host-B:161 Community="public", Oid="1.3.6.1.2.1.1.1.0"
            var requestId = Program.RequestCounter.NextId;
            var message = new GetRequestMessage(requestId, VersionCode.V1, Program.Community, new List<Variable> { new Variable(Program.Oid) });

            using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10))) // timeout 10 sec
            using (var client = new SnmpClient())
            {
                try
                {
                    var response = await client.GetResponseAsync(message, Program.RemoteEp, cts.Token);
                    var responseVariablesText = string.Join(
                        ", ",
                        response.Variables()
                            .Select(variable => string.Format("[OID={0} Data={1}]", variable.Id, variable.Data)));
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, responseVariablesText);
                }
                catch (OperationCanceledException ex)
                {
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, ex.Message);
                }
                catch (Exception ex)
                {
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, ex.Message);
                }
                finally
                {
                }
            }
        }

        #endregion

        #region Test2 (Extension Method sample)

        private static async Task PrepareTestMemory2()
        {
            await Program.TestToCallSnmpClient2(-1);
        }
        private static async Task TestMemory2()
        {
            // Running this code on Host-A
            // The Snmp Agent on Host-B, but an agent is not listening...

            var tasks = Enumerable.Range(0, Program.TestCount).Select(async (_, index) => await Program.TestToCallSnmpClient2(index));
            var allTasks = Task.WhenAll(tasks);
            try
            {
                await allTasks.ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Console.WriteLine("{0}", ex.InnerException);
            }
        }
        private static async Task TestToCallSnmpClient2(int index)
        {
            // request: RemoteEP=Host-B:161 Community="public", Oid="1.3.6.1.2.1.1.1.0"
            var requestId = Program.RequestCounter.NextId;
            var message = new GetRequestMessage(requestId, VersionCode.V1, Program.Community, new List<Variable> { new Variable(Program.Oid) });

            using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10))) // timeout 10 sec
            {
                try
                {
                    var response = await message.GetResponseAsync(Program.RemoteEp, cts.Token);
                    var responseVariablesText = string.Join(
                        ", ",
                        response.Variables()
                            .Select(variable => string.Format("[OID={0} Data={1}]", variable.Id, variable.Data)));
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, responseVariablesText);
                }
                catch (OperationCanceledException ex)
                {
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, ex.Message);
                }
                catch (Exception ex)
                {
                    Console.WriteLine("{0:d3}({1}):{2}", index, requestId, ex.Message);
                }
                finally
                {
                }
            }
        }

        #endregion

        #region Extension Methods

        internal static async Task<T> WithCancellation<T>(this Task<T> task, CancellationToken cancellationToken)
        {
            var tcs = new TaskCompletionSource<bool>();
            using (cancellationToken.Register(state => ((TaskCompletionSource<bool>)state).TrySetResult(true), tcs))
            {
                var completedTask = await Task.WhenAny(task, tcs.Task).ConfigureAwait(false);
                if (task != completedTask)
                {
                    throw new OperationCanceledException(cancellationToken);
                }
            }

            return await task.ConfigureAwait(false);
        }

        //public static IDisposable CreateTimeoutScope(this IDisposable disposable, TimeSpan timeSpan)
        //{
        //    var cancellationTokenSource = new CancellationTokenSource(timeSpan);
        //    var cancellationTokenRegistration = cancellationTokenSource.Token.Register(disposable.Dispose);
        //    return new DisposableScope(
        //        () =>
        //        {
        //            cancellationTokenRegistration.Dispose();
        //            cancellationTokenSource.Dispose();
        //            disposable.Dispose();
        //        });
        //}

        //private sealed class DisposableScope : IDisposable
        //{
        //    private readonly Action closeScopeAction;
        //    public DisposableScope(Action closeScopeAction)
        //    {
        //        this.closeScopeAction = closeScopeAction;
        //    }
        //    public void Dispose()
        //    {
        //        this.closeScopeAction();
        //    }
        //}

        #endregion

        #region SnmpClient for sample

        private class SnmpClient : IDisposable
        {
            private readonly CancellationTokenSource cts;
            private long disposableState;

            public SnmpClient()
            {
                this.cts = new CancellationTokenSource();
            }

            ~SnmpClient()
            {
                this.Dispose(false);
            }

            public bool IsDisposed
            {
                get { return Interlocked.Read(ref this.disposableState) == 1L; }
            }

            public void Dispose()
            {
                this.Dispose(true);
                GC.SuppressFinalize(this);
            }

            protected virtual void DisposeManagedInstances()
            {
                this.cts.Cancel();
                this.cts.Dispose();
            }

            protected virtual void DisposeUnmanagedInstances()
            {
            }

            private void Dispose(bool disposing)
            {
                if (Interlocked.CompareExchange(ref this.disposableState, 1L, 0L) == 0L)
                {
                    Debug.Print("!! {0}.Dispose...", this.GetType().FullName);
                    if (disposing)
                    {
                        this.DisposeManagedInstances();
                    }

                    this.DisposeUnmanagedInstances();
                }
            }

            public async Task<ISnmpMessage> GetResponseAsync(ISnmpMessage request, IPEndPoint remoteEndPoint, CancellationToken cancellationToken)
            {
                using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(this.cts.Token, cancellationToken))
                using (var client = UdpClientWrapper.CreateClient())
                {
                    var sendingBytes = request.ToBytes();
                    var sendingTask = client.SendAsync(sendingBytes, sendingBytes.Length, remoteEndPoint);
                    var sentBytesSize = await sendingTask.WithCancellation(linkedCts.Token).ConfigureAwait(false);

                    var receivingTask = client.ReceiveAsync();
                    var receiveResult = await receivingTask.WithCancellation(linkedCts.Token).ConfigureAwait(false);

                    var reply = receiveResult.Buffer;
                    var registry = new UserRegistry();
                    var response = MessageFactory.ParseMessages(reply, 0, receiveResult.Buffer.Length, registry)[0];

                    var responseCode = response.TypeCode();
                    if (responseCode == SnmpType.ResponsePdu || responseCode == SnmpType.ReportPdu)
                    {
                        var requestId = request.MessageId();
                        var responseId = response.MessageId();
                        if (responseId != requestId)
                        {
                            throw OperationException.Create(string.Format(CultureInfo.InvariantCulture, "wrong response sequence: expected {0}, received {1}", requestId, responseId), remoteEndPoint.Address);
                        }

                        return response;
                    }

                    throw OperationException.Create(string.Format(CultureInfo.InvariantCulture, "wrong response type: {0}", responseCode), remoteEndPoint.Address);
                }
            }
        }

        #endregion

        #region UdpClientWrapper for sample

        private class UdpClientWrapper : UdpClient
        {
            private UdpClientWrapper()
                : base(AddressFamily.InterNetwork)
            {
            }

            public static UdpClientWrapper CreateClient()
            {
                return new UdpClientWrapper();
            }

            protected override void Dispose(bool disposing)
            {
                Debug.Print("!! {0}.Dispose...", this.GetType().FullName);
                base.Dispose(disposing);
            }
        }

        #endregion

        #region Extension Method version for sample

        private static async Task<ISnmpMessage> GetResponseAsync(this ISnmpMessage request, IPEndPoint remoteEndPoint, CancellationToken cancellationToken)
        {
            using (var client = UdpClientWrapper.CreateClient())
            {
                var sendingBytes = request.ToBytes();
                var sendingTask = client.SendAsync(sendingBytes, sendingBytes.Length, remoteEndPoint);
                var sentBytesSize = await sendingTask.WithCancellation(cancellationToken).ConfigureAwait(false);

                var receivingTask = client.ReceiveAsync();
                var receiveResult = await receivingTask.WithCancellation(cancellationToken).ConfigureAwait(false);

                var reply = receiveResult.Buffer;
                var registry = new UserRegistry();
                var response = MessageFactory.ParseMessages(reply, 0, receiveResult.Buffer.Length, registry)[0];

                var responseCode = response.TypeCode();
                if (responseCode == SnmpType.ResponsePdu || responseCode == SnmpType.ReportPdu)
                {
                    var requestId = request.MessageId();
                    var responseId = response.MessageId();
                    if (responseId != requestId)
                    {
                        throw OperationException.Create(string.Format(CultureInfo.InvariantCulture, "wrong response sequence: expected {0}, received {1}", requestId, responseId), remoteEndPoint.Address);
                    }

                    return response;
                }

                throw OperationException.Create(string.Format(CultureInfo.InvariantCulture, "wrong response type: {0}", responseCode), remoteEndPoint.Address);
            }
        }

        #endregion
    }
}

SampleByUdpClient.md

Sample by UdpClient