Ticket #113 : Add documentation on AMQP1.0 plugin

Author: habarthierry-hue

FaasNet.Samples.sln

@@ -0,0 +1,30 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 17
+VisualStudioVersion = 17.1.32328.378
+MinimumVisualStudioVersion = 10.0.40219.1
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "01. EventMesh", "01. EventMesh", "{7A371086-78D9-4632-96F4-DCF31581F3AE}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "FaasNet.EventMesh.AMQPClient", "src\Samples\FaasNet.EventMesh.AMQPClient\FaasNet.EventMesh.AMQPClient.csproj", "{6AE025D4-3488-4809-A370-5682A30E820D}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{6AE025D4-3488-4809-A370-5682A30E820D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6AE025D4-3488-4809-A370-5682A30E820D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6AE025D4-3488-4809-A370-5682A30E820D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6AE025D4-3488-4809-A370-5682A30E820D}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(NestedProjects) = preSolution
+		{6AE025D4-3488-4809-A370-5682A30E820D} = {7A371086-78D9-4632-96F4-DCF31581F3AE}
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {007D7A77-3286-48F7-974F-20075BCB6160}
+	EndGlobalSection
+EndGlobal

docs/documentation/eventmesh/pluginamqp.md

@@ -26,4 +26,110 @@ The ZIP file can be downloaded [here]().
 
 ## Quick start
 
-TODO
+Once you have an up and running EventMesh server with `ProtocolAmqp` plugin enabled, you can start using any client compliant with the AMQP 1.0 protocol.
+
+### Configure client and VPN
+
+Before going further, a Virtual Private Network (VPN) and two clients must be configured.
+Those information will be used to publish message and subscribe to one or more topic.
+
+Open a command prompt and create a topic named `default` :
+
+```
+FaasNet.EventMeshCTL.CLI.exe add_vpn --name=default
+```
+
+Add a client `publishClient`, as the name suggests, it will be used to publish message.
+
+```
+FaasNet.EventMeshCTL.CLI.exe add_client --vpn=default --identifier=publishClient --publish_enabled=true --subscription_enabled=false
+```
+
+Add a client `subscribeClient`, this client will be used for subscription.
+
+```
+FaasNet.EventMeshCTL.CLI.exe add_client --vpn=default --identifier=subscribeClient --publish_enabled=false --subscription_enabled=true
+```
+
+### Configure the plugin
+
+If the plugin is not yet configured, it can be enabled like this
+
+```
+FaasNet.EventMeshCTL.CLI.exe get_plugin_configuration --name=ProtocolAmqp
+```
+
+Its configuration can be updated either by [using CLI](cli.md) or by updating the configuration file `appsettings.json`.
+
+Don't forget that the EventMesh server must be restarted, otherwise the changes are not taken into account.
+
+When the configuration is finished, a client can be created and can start publishing message.
+
+## Source Code
+
+The source code of this project can be found [here]().
+
+### Create a client
+
+In this tutorial, we will explain how to create a C# client to publish message and subscribe to one topic `q1`.
+
+Create a new console application and install the Nuget package `AmqpNetLite`
+
+```
+dotnet add package AmqpNetLite 
+```
+
+In the `Program.cs` file add a static method used to send message.
+
+```
+static void SendMessage()
+{
+    var address = new Address($"amqp://publishClient:publishClient@localhost:{port}");
+    var connection = new Connection(address);
+    var session = new Session(connection);
+    var message = new Message("Hello AMQP!");
+    var sender = new SenderLink(session, "sender-link", "q1");
+    Console.WriteLine("Sent Hello AMQP!");
+    sender.Send(message);
+    sender.Close();
+    session.Close();
+    connection.Close();
+}
+```
+
+Add a second static method used to subscribe to a topic :
+
+```
+static void ReceiveMessage()
+{
+    Task.Run(() =>
+    {
+        var address = new Address($"amqp://subscribeClient:subscribeClient@localhost:{port}");
+        var connection = new Connection(address);
+        var session = new Session(connection);
+        var receiver = new ReceiverLink(session, "receiver-link", "q1");
+        var message = receiver.Receive();
+        Console.WriteLine("Received " + message.Body.ToString());
+    });
+}
+```
+
+Call both methods like this :
+
+```
+const int port = 5672;
+Console.WriteLine("Hello, World!");
+ReceiveMessage();
+SendMessage();
+Console.WriteLine("Press enter to quit the application");
+Console.ReadLine();
+```
+
+Build the project and run.
+
+The console application should displayed two messages :
+
+```
+Send Hello AMQP!
+Received Hello AMQP!
+```

docs/documentation/eventmesh/plugins.md

@@ -41,7 +41,7 @@ FaasNet.EventMeshCTL.CLI.exe update_plugin_configuration --name=<PLUGIN_NAME> --
 ```
 
 > [!IMPORTANT]
-> Once the plugin is properly configured. The EventMesh server must be restarted in order to take into consideration your modifications.
+> Once the plugin is properly configured. The EventMesh server must be restarted in order to take into account your modifications.
 
 ## Supported plugins
 

src/EventMesh/FaasNet.EventMesh.Plugin/PluginEntryDiscovery.cs

@@ -11,17 +11,23 @@ namespace FaasNet.EventMesh.Plugin
 {
     public class PluginEntryDiscovery
     {
-        public static bool TryExtract(string pluginDirectoryPath, IEnumerable<string> activePlugins, out IDiscoveredPlugin discoveryPlugin)
+        public static bool TryExtract(string pluginDirectoryPath, IEnumerable<string> activePlugins, IEnumerable<Type> additionalSharedTypes, out IDiscoveredPlugin discoveryPlugin)
         {
             discoveryPlugin = null;
+            var sharedTypes = new List<Type> { typeof(IPlugin<>), typeof(IServiceCollection) };
+            if(additionalSharedTypes != null)
+            {
+                foreach(var additionalSharedType in additionalSharedTypes) sharedTypes.Add(additionalSharedType);
+            }
+
             var pluginEntry = PluginConfigurationFile.Read(pluginDirectoryPath);
             if(pluginEntry == null) return false;
             if (!activePlugins.Contains(pluginEntry.Name)) return false;
             var dllPath = Path.Combine(pluginDirectoryPath, pluginEntry.DllName);
             if (!File.Exists(dllPath)) return false;
             var loader = PluginLoader.CreateFromAssemblyFile(
                 dllPath,
-                sharedTypes: new[] { typeof(IPlugin<>), typeof(IServiceCollection) });
+                sharedTypes: sharedTypes.ToArray());
             var assembly = loader.LoadDefaultAssembly();
             var types = assembly.GetTypes();
             var pluginType = types.FirstOrDefault(t => t.GetInterfaces().Any(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IPlugin<>)));

src/EventMesh/FaasNet.EventMesh.Runtime/Handlers/PublishMessageRequestHandler.cs

@@ -86,6 +86,7 @@ private async Task SendMessage(PublishMessageRequest message, CancellationToken
         private async Task<ClusterNode> GetRandomClusterNode(CancellationToken cancellationToken)
         {
             var nodes = await _clusterStore.GetAllNodes(cancellationToken);
+            if (nodes.Count() == 1) return nodes.First();
             nodes = nodes.Where(n => n.Port != _nodeOptions.Port || n.Url != _nodeOptions.Url);
             var rnd = new Random();
             var rndIndex = rnd.Next(0, nodes.Count() - 1);

src/EventMesh/FaasNet.EventMesh.Service/EventMeshProxyWorker.cs

@@ -0,0 +1,25 @@
+using FaasNet.EventMesh.Protocols;
+
+namespace FaasNet.EventMesh.Service
+{
+    public class EventMeshProxyWorker : IHostedService
+    {
+        private readonly IEnumerable<IProxy> _proxyLst;
+
+        public EventMeshProxyWorker(IEnumerable<IProxy> proxyLst)
+        {
+            _proxyLst = proxyLst;
+        }
+
+        public async Task StartAsync(CancellationToken cancellationToken)
+        {
+            foreach (var proxy in _proxyLst) await proxy.Start();
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            foreach (var proxy in _proxyLst) proxy.Stop();
+            return Task.CompletedTask;
+        }
+    }
+}

src/EventMesh/FaasNet.EventMesh.Service/Program.cs

@@ -1,8 +1,10 @@
 using FaasNet.Common;
 using FaasNet.EventMesh.Plugin;
+using FaasNet.EventMesh.Protocols;
 using FaasNet.EventMesh.Runtime;
 using FaasNet.EventMesh.Runtime.Stores;
 using FaasNet.EventMesh.Service;
+using FaasNet.EventMesh.Sink;
 
 using IHost host = Host.CreateDefaultBuilder(args)
                 .UseWindowsService(o =>
@@ -26,6 +28,7 @@
                     var activePlugins = pluginStore.GetActivePlugins();
                     LoadProtocolPlugins(serverBuilder.Services, activePlugins);
                     LoadSinkPlugins(serverBuilder.Services, activePlugins);
+                    services.AddHostedService<EventMeshServerWorker>();
                 }).Build();
 await host.RunAsync();
 
@@ -36,11 +39,11 @@ static void LoadProtocolPlugins(IServiceCollection services, IEnumerable<string>
     if (Directory.Exists(pluginsDirectory)) pluginPaths = Directory.EnumerateDirectories(pluginsDirectory);
     var discoveredProtocolPlugins = pluginPaths.Select(p =>
     {
-        if (PluginEntryDiscovery.TryExtract(p, activePlugins, out IDiscoveredPlugin discoveredPlugin)) return discoveredPlugin;
+        if (PluginEntryDiscovery.TryExtract(p, activePlugins, new[] { typeof(IProxy) }, out IDiscoveredPlugin discoveredPlugin)) return discoveredPlugin;
         return null;
     }).Where(p => p != null);
     foreach (var discoveredPlugin in discoveredProtocolPlugins) discoveredPlugin.Load(services);
-    services.AddHostedService<EventMeshServerWorker>();
+    services.AddHostedService<EventMeshProxyWorker>();
 }
 
 static void LoadSinkPlugins(IServiceCollection services, IEnumerable<string> activePlugins)
@@ -50,7 +53,7 @@ static void LoadSinkPlugins(IServiceCollection services, IEnumerable<string> act
     if (Directory.Exists(pluginsDirectory)) pluginPaths = Directory.EnumerateDirectories(pluginsDirectory);
     var discoveredSinkPlugins = pluginPaths.Select(p =>
     {
-        if (PluginEntryDiscovery.TryExtract(p, activePlugins, out IDiscoveredPlugin discoveredPlugin)) return discoveredPlugin;
+        if (PluginEntryDiscovery.TryExtract(p, activePlugins, new[] { typeof(ISinkJob) }, out IDiscoveredPlugin discoveredPlugin)) return discoveredPlugin;
         return null;
     }).Where(p => p != null);
     foreach (var discoveredPlugin in discoveredSinkPlugins) discoveredPlugin.Load(services);

src/EventMesh/FaasNet.EventMeshCTL.CLI/Program.cs

@@ -84,7 +84,7 @@ private static void AddVpnCommand(CommandLineApplication app)
                         return;
                     }
 
-                    Console.WriteLine($"VPN {vpnName} has been added");
+                    DisplaySuccess($"VPN {vpnName} has been added");
                 });
             });
         }
@@ -115,7 +115,7 @@ private static void AddClientCommand(CommandLineApplication app)
                 idOption.DefaultValue = "clientId";
                 var pubEnabled = addClientCmd.Option<bool>("-p|--publish_enabled <PUBLISH>", "Enable publish", CommandOptionType.SingleValue);
                 pubEnabled.DefaultValue = true;
-                var subEnabled = addClientCmd.Option<bool>("-s|--subription_enabled <SUBSCRIPTION>", "Enable subscription", CommandOptionType.SingleValue);
+                var subEnabled = addClientCmd.Option<bool>("-s|--subscription_enabled <SUBSCRIPTION>", "Enable subscription", CommandOptionType.SingleValue);
                 subEnabled.DefaultValue = true;
                 addClientCmd.OnExecuteAsync(async (token) =>
                 {
@@ -134,7 +134,7 @@ private static void AddClientCommand(CommandLineApplication app)
                         return;
                     }
 
-                    Console.WriteLine($"Client {idOption.ParsedValue} has been added");
+                    DisplaySuccess($"Client {idOption.ParsedValue} has been added");
                 });
             });
         }

src/RaftConsensus/FaasNet.RaftConsensus.Core/BasePeerHost.cs

@@ -183,12 +183,14 @@ private async Task StartElection()
             _logger.LogInformation("{Node}:{PeerId}:{TermId}, Start to vote", _nodeId, _peerId, Info.TermId);
             var nodes = await _clusterStore.GetAllNodes(TokenSource.Token);
             nodes = nodes.Where(n => n.Port != _nodeOptions.Port || n.Url != _nodeOptions.Url);
-            _quorum = (nodes.Count() / 2) + 1;
+            _quorum = !nodes.Any() ? 0 : (nodes.Count() / 2) + 1;
+            /*
             if (_quorum == 0)
             {
                 _logger.LogError("{Node}:{PeerId}:{TermId}, There is no enough nodes", _nodeId, _peerId, Info.TermId);
                 return;
             }
+            */
 
             _nbPositiveVote = 0;
             _expirationCheckElectionDateTime = DateTime.UtcNow.AddMilliseconds(_options.ElectionCheckDurationMS);

src/Samples/FaasNet.EventMesh.AMQPClient/FaasNet.EventMesh.AMQPClient.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net6.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="AmqpNetLite" Version="2.4.4" />
+  </ItemGroup>
+
+</Project>