From bdcef9d53436046b7cf252696b25a4b7ecd4f9e0 Mon Sep 17 00:00:00 2001
From: David Jensen <djn@danskecommodities.com>
Date: Fri, 12 Apr 2024 12:46:13 +0200
Subject: [PATCH] Add PulsarService and functionality to PulsarBuilder

A new file, PulsarService.cs, has been added, with a struct named PulsarService that allows the creation of different Pulsar services. Changes have also been made to PulsarBuilder.cs, mainly to incorporate the use of PulsarService. Rather than using a constant value, `WithAuthentication()` and `WithFunctionsWorker(bool functionsWorkerEnabled = true)` now use elements from the PulsarService struct. Wait strategy has been updated to better account for various enabled services.
---
 src/Testcontainers.Pulsar/PulsarBuilder.cs | 46 +++++++++++++++-------
 src/Testcontainers.Pulsar/PulsarService.cs | 30 ++++++++++++++
 2 files changed, 61 insertions(+), 15 deletions(-)
 create mode 100644 src/Testcontainers.Pulsar/PulsarService.cs

diff --git a/src/Testcontainers.Pulsar/PulsarBuilder.cs b/src/Testcontainers.Pulsar/PulsarBuilder.cs
index f5445f740..4d875fe56 100644
--- a/src/Testcontainers.Pulsar/PulsarBuilder.cs
+++ b/src/Testcontainers.Pulsar/PulsarBuilder.cs
@@ -1,3 +1,5 @@
+using System.Linq;
+
 namespace Testcontainers.Pulsar;
 
 /// <inheritdoc cref="ContainerBuilder{TBuilderEntity, TContainerEntity, TConfigurationEntity}" />
@@ -17,6 +19,8 @@ public sealed class PulsarBuilder : ContainerBuilder<PulsarBuilder, PulsarContai
     public const string Username = "test-user";
 
     private static readonly IReadOnlyDictionary<string, string> AuthenticationEnvVars;
+    
+    private static readonly ISet<PulsarService> EnabledServices = new HashSet<PulsarService>();
 
     static PulsarBuilder()
     {
@@ -58,22 +62,27 @@ private PulsarBuilder(PulsarConfiguration resourceConfiguration)
     protected override PulsarConfiguration DockerResourceConfiguration { get; }
 
     /// <summary>
-    ///
+    /// Enables authentication in the Pulsar container.
     /// </summary>
-    /// <returns></returns>
+    /// <returns>A <see cref="PulsarBuilder"/> instance.</returns>
     public PulsarBuilder WithAuthentication()
     {
+        EnabledServices.Add(PulsarService.Authentication);
+        
         return Merge(DockerResourceConfiguration, new PulsarConfiguration(authenticationEnabled: true))
             .WithEnvironment(AuthenticationEnvVars);
     }
 
     /// <summary>
-    ///
+    /// Initializes a new instance of the <see cref="WithFunctionsWorkerWorker" /> class.
     /// </summary>
-    /// <returns></returns>
+    /// <param name="functionsWorkerEnabled">Determines if the functions worker is enabled.</param>
+    /// <returns>A new instance of the <see cref="PulsarBuilder" /> class.</returns>
     public PulsarBuilder WithFunctionsWorker(bool functionsWorkerEnabled = true)
     {
-        // TODO: When enabled we need to adjust the wait strategy.
+        if (functionsWorkerEnabled)
+            EnabledServices.Add(PulsarService.FunctionWorker);
+        
         return Merge(DockerResourceConfiguration, new PulsarConfiguration(functionsWorkerEnabled: functionsWorkerEnabled));
     }
 
@@ -84,18 +93,25 @@ public override PulsarContainer Build()
         
         var waitStrategy = Wait.ForUnixContainer();
 
-        //TODO We need to switch between the default and custom WaitStrategy depending on if the user used WithAuthentication.
-        //Would you prefer we handled it in a similar to Couchbase?
-        waitStrategy = waitStrategy.UntilHttpRequestIsSucceeded(request
-            => request
-                .ForPath("/admin/v2/clusters")
-                .ForPort(PulsarWebServicePort)
-                .ForResponseMessageMatching(VerifyPulsarStatusAsync));
-        
-        waitStrategy.AddCustomWaitStrategy(new WaitUntil());
+        if (EnabledServices.Contains(PulsarService.Authentication))
+        {
+            waitStrategy.AddCustomWaitStrategy(new WaitUntil());
+        }
+        else
+        {
+            waitStrategy = waitStrategy.UntilHttpRequestIsSucceeded(request
+                => request
+                    .ForPath("/admin/v2/clusters")
+                    .ForPort(PulsarWebServicePort)
+                    .ForResponseMessageMatching(VerifyPulsarStatusAsync));
+        }
         
-        var pulsarBuilder = WithWaitStrategy(waitStrategy);
+        if (EnabledServices.Contains(PulsarService.FunctionWorker))
+        {
+            waitStrategy.UntilMessageIsLogged(".*Function worker service started.*");
+        }
         
+        var pulsarBuilder =  WithWaitStrategy(waitStrategy);
         return new PulsarContainer(pulsarBuilder.DockerResourceConfiguration);
     }
 
diff --git a/src/Testcontainers.Pulsar/PulsarService.cs b/src/Testcontainers.Pulsar/PulsarService.cs
new file mode 100644
index 000000000..f9dc3c716
--- /dev/null
+++ b/src/Testcontainers.Pulsar/PulsarService.cs
@@ -0,0 +1,30 @@
+namespace Testcontainers.Pulsar;
+
+internal readonly struct PulsarService
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="PulsarService" /> struct.
+    /// </summary>
+    /// <param name="identifier">The identifier.</param>
+    private PulsarService(string identifier)
+    {
+        Identifier = identifier;
+    }
+    
+    /// <summary>
+    /// Gets the Auth.
+    /// </summary>
+    public static readonly PulsarService Authentication = new PulsarService("auth");
+    
+    /// <summary>
+    /// Gets the Function worker service.
+    /// </summary>
+    public static readonly PulsarService FunctionWorker = new PulsarService("funk");
+    
+    /// <summary>
+    /// Gets the identifier.
+    /// </summary>
+    public string Identifier { get; }
+}
+
+