Création de Kinesis Agent pour les plugins Windows

Créez un projet de bibliothèque C # ciblant .NET Framework 4.6.

Ajoutez une référence à l'assembly Amazon.KinesisTap.Core. Cet assembly est situé dans le%PROGRAMFILES%\Amazon\AWSKinesisTapaprès l'installation de Kinesis Agent pour Windows.

Utilisez NuGet pour installer le package Microsoft.Extensions.Configuration.Abstractions.

Utilisez NuGet pour installer le package System.Reactive.

Utilisez NuGet pour installer le package Microsoft.Extensions.Logging.

Créez une classe Factory qui implémente IFactory<IEventSource> pour les sources ou IFactory<IEventSink> pour les récepteurs. Ajoutez les méthodes RegisterFactory et CreateInstance.

Par exemple, le code suivant crée une fabrique de plug-ins Kinesis Agent pour Windows qui crée une source générant des données aléatoires :

using System;
using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Configuration;

namespace MyCompany.MySources
{
    public class RandomSourceFactory : IFactory<ISource>
    {
        public void RegisterFactory(IFactoryCatalog<ISource> catalog)
        {
            catalog.RegisterFactory("randomsource", this);
        }

        public ISource CreateInstance(string entry, IPlugInContext context)
        {
            IConfiguration config = context.Configuration;

            switch (entry.ToLower())
            {
                case "randomsource":
                    string rateString = config["Rate"];
                    string maxString = config["Max"];
                    TimeSpan rate;
                    int max;

                    if (string.IsNullOrWhiteSpace(rateString))
                    {
                        rate = TimeSpan.FromSeconds(30);
                    }
                    else
                    {
                        if (!TimeSpan.TryParse(rateString, out rate))
                        {
                            throw new Exception($"Rate {rateString} is invalid for RandomSource.");
                        }
                    }
 
                    if (string.IsNullOrWhiteSpace(maxString))
                    {
                        max = 1000;
                    }
                    else
                    {
                        if (!int.TryParse(maxString, out max))
                        {
                            throw new Exception($"Max {maxString} is invalid for RandomSource.");
                        }
                    }

                    return new RandomSource(rate, max, context);
                default:
                    throw new ArgumentException($"Source {entry} is not recognized.", entry);
            }
        }
    }
}               

L'instruction switch est utilisée dans la méthode CreateInstance au cas où vous souhaiteriez améliorer la fabrique de façon à créer différents types d'instances.

Pour créer une fabrique de récepteurs qui crée un récepteur qui ne fait rien, utilisez une classe similaire à la classe suivante :

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Configuration;

namespace MyCompany.MySinks 
{
    public class NullSinkFactory : IFactory<IEventSink>
    {
        public void RegisterFactory(IFactoryCatalog<IEventSink> catalog)
        {
            catalog.RegisterFactory("nullsink", this);
        }

        public IEventSink CreateInstance(string entry, IPlugInContext context)
        {
            IConfiguration config = context.Configuration;

            switch (entry.ToLower())
            {
                case "nullsink":
                    return new NullSink(context);
                default:
                    throw new Exception("Unrecognized sink type {entry}.");
            }
        }
    }
}             

Ajoutez une classe qui implémente l'interface IEventSource<out T> au projet créé précédemment pour la source.

Par exemple, utilisez le code suivant pour définir une source qui génère des données aléatoires :

using System;
using System.Reactive.Subjects;
using System.Timers;
using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Logging;

namespace MyCompany.MySources
{
    public class RandomSource : EventSource<RandomData>, IDisposable
    {
        private TimeSpan _rate;
        private int _max;
        private Timer _timer = null;
        private Random _random = new Random();
        private ISubject<IEnvelope<RandomData>> _recordSubject = new Subject<IEnvelope<RandomData>>();
        


        public RandomSource(TimeSpan rate, int max, IPlugInContext context) : base(context)
        {
            _rate = rate;
            _max = max;
        }

        public override void Start()
        {
            try
            {
                CleanupTimer();
                _timer = new Timer(_rate.TotalMilliseconds);
                _timer.Elapsed += (Object source, ElapsedEventArgs args) =>
                {
                    var data = new RandomData()
                    {
                        RandomValue = _random.Next(_max)
                    };
                    _recordSubject.OnNext(new Envelope<RandomData>(data));
                };
                _timer.AutoReset = true;
                _timer.Enabled = true;
                _logger?.LogInformation($"Random source id {this.Id} started with rate {_rate.TotalMilliseconds}.");
            }
            catch (Exception e)
            {
                _logger?.LogError($"Exception during start of RandomSource id {this.Id}: {e}");
            }
        }

        public override void Stop()
        {
            try
            {
                CleanupTimer();
                _logger?.LogInformation($"Random source id {this.Id} stopped.");                
            }
            catch (Exception e)
            {
                _logger?.LogError($"Exception during stop of RandomSource id {this.Id}: {e}");
            }
        }

        private void CleanupTimer()
        {
            if (_timer != null)
            {
                _timer.Enabled = false;
                _timer?.Dispose();
                _timer = null;
            }
        }

        public override IDisposable Subscribe(IObserver<IEnvelope<RandomData>> observer)
        {
            return this._recordSubject.Subscribe(observer);
        }

        public void Dispose()
        {
            CleanupTimer();
        }
    }
}               

Dans cet exemple, la classe RandomSource hérite de la classe EventSource<T> étant donné qu'elle fournit la propriété Id. Bien que cet exemple ne prenne pas en charge l'attribution de signets, cette classe de base est également utile pour implémenter cette fonctionnalité. Les enveloppes permettent de stocker des métadonnées et d'envelopper les données arbitraires pour les diffuser auprès des récepteurs. La classe RandomData est définie à l'étape suivante et représente le type d'objet de sortie de cette source.

Ajoutez une classe au projet défini précédemment, qui contient les données qui sont diffusées à partir de la source.

Par exemple, un conteneur de données aléatoires peut être défini comme suit :

namespace MyCompany.MySources
{
    public class RandomData
    {
        public int RandomValue { get; set; }
    }
}               

Compilez le projet précédemment défini.

Copiez l'assembly dans le répertoire d'installation de Kinesis Agent for Windows.

Créez ou mettez à jour unappsettings.jsonqui utilise la nouvelle source et placez-le dans le répertoire d'installation de Kinesis Agent pour Windows.

Arrêtez et démarrez Kinesis Agent pour Windows.

Vérifiez le fichier journal Kinesis Agent pour Windows actuel (généralement situé dans le répertoire%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) afin de vous assurer qu'il n'y a aucun problème avec le plug-in source personnalisé.

Vérifiez que les données sont reçues par le service AWS souhaité.

Ajoutez une classe à l'objet défini précédemment qui implémente l'interface IEventSink.

Par exemple, le code suivant implémente un récepteur qui ne fait que consigner l'arrivée d'enregistrements, qui sont ensuite supprimés.

using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Logging;

namespace MyCompany.MySinks
{
    public class NullSink : EventSink
    {
        public NullSink(IPlugInContext context) : base(context)
        {
        }

        public override void OnNext(IEnvelope envelope)
        {
            _logger.LogInformation($"Null sink {Id} received {GetRecord(envelope)}.");
        }

        public override void Start()
        {
            _logger.LogInformation($"Null sink {Id} starting.");
        }

        public override void Stop()
        {
            _logger.LogInformation($"Null sink {Id} stopped.");
        }
    }
}               

Dans cet exemple, la classe de récepteur NullSink hérite de la classe EventSink, car elle permet de transformer des enregistrements en différents formats de sérialisation tels que JSON et XML.

Compilez le projet précédemment défini.

Copiez l'assembly dans le répertoire d'installation de Kinesis Agent for Windows.

Créez ou mettez à jour unappsettings.jsonqui utilise le nouveau récepteur et placez-le dans le répertoire d'installation de Kinesis Agent pour Windows. Par exemple, pour utiliser les plug-ins personnalisés RandomSource et NullSink, vous pouvez utiliser le fichier de configuration appsettings.json suivant :

{
  "Sources": [
  {
	"Id": "MyRandomSource",
	"SourceType": "RandomSource",
	"Rate": "00:00:10",
	"Max": 50
  }
    
  ],
  "Sinks": [
  {
    "Id": "MyNullSink",
	"SinkType": "NullSink",
	"Format": "json"
  }
  ],
  "Pipes": [
    {
	  "Id": "MyRandomToNullPipe",
	  "SourceRef": "MyRandomSource",
	  "SinkRef": "MyNullSink"
	}
  ]
}               

Cette configuration crée une source qui envoie une instance de RandomData avec la valeur RandomValue définie sur un nombre aléatoire compris entre 0 et 50 toutes les 10 secondes. Cela crée un récepteur qui transforme les instances RandomData entrantes au format JSON, enregistre ce JSON, puis supprime les instances. Assurez-vous d'inclure les deux exemples de fabriques, la classe de source RandomSource et la classe de récepteur NullSink du projet défini précédemment pour utiliser l'exemple de fichier de configuration.

Arrêtez et démarrez Kinesis Agent pour Windows.

Vérifiez le fichier journal Kinesis Agent pour Windows actuel (généralement situé dans le répertoire%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) afin de vous assurer qu'il n'y a aucun problème avec le plug-in de récepteur personnalisé.

Vérifiez que les données sont reçues par le service AWS souhaité. Étant donné que l'exemple de classe NullSink ne diffuse vers aucun service AWS, vous pouvez vérifier le bon fonctionnement du récepteur en recherchant les messages de journaux indiquant que les enregistrements ont été reçus.

Par exemple, vous pouvez accéder à un fichier journal semblable à :

2018-10-18 12:36:36.3647 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.AWS.AWSEventSinkFactory.
2018-10-18 12:36:36.4018 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Windows.PerformanceCounterSinkFactory.
2018-10-18 12:36:36.4018 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory MyCompany.MySinks.NullSinkFactory.
2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Core.DirectorySourceFactory.
2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.ExchangeSource.ExchangeSourceFactory.
2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Uls.UlsSourceFactory.
2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Windows.WindowsSourceFactory.
2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory MyCompany.MySources.RandomSourceFactory.
2018-10-18 12:36:36.9601 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Core.Pipes.PipeFactory.
2018-10-18 12:36:37.4694 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.AutoUpdate.AutoUpdateFactory.
2018-10-18 12:36:37.4807 Amazon.KinesisTap.Hosting.LogManager INFO Performance counter sink  started.
2018-10-18 12:36:37.6250 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink starting.
2018-10-18 12:36:37.6250 Amazon.KinesisTap.Hosting.LogManager INFO Connected source MyRandomSource to sink MyNullSink
2018-10-18 12:36:37.6333 Amazon.KinesisTap.Hosting.LogManager INFO Random source id MyRandomSource started with rate 10000.
2018-10-18 12:36:47.8084 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":14}.
2018-10-18 12:36:57.6339 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":5}.
2018-10-18 12:37:07.6490 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":9}.
2018-10-18 12:37:17.6494 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":47}.
2018-10-18 12:37:27.6520 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":25}.
2018-10-18 12:37:37.6676 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":21}.
2018-10-18 12:37:47.6688 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":29}.
2018-10-18 12:37:57.6700 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":22}.
2018-10-18 12:38:07.6838 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":32}.
2018-10-18 12:38:17.6848 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":12}.
2018-10-18 12:38:27.6866 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":46}.
2018-10-18 12:38:37.6880 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":48}.
2018-10-18 12:38:47.6893 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":39}.
2018-10-18 12:38:57.6906 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":18}.
2018-10-18 12:39:07.6995 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":6}.
2018-10-18 12:39:17.7004 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":0}.
2018-10-18 12:39:27.7021 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":3}.
2018-10-18 12:39:37.7023 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":19}.