From c4402bed0964f0a55abdb8d785cccc2b5c7abebc Mon Sep 17 00:00:00 2001
From: David Jensen <djn@danskecommodities.com>
Date: Fri, 12 Apr 2024 10:32:01 +0200
Subject: [PATCH] Update Pulsar image and enhance wait strategy

The Pulsar image has been upgraded from apachepulsar/pulsar:3.0.3 to apachepulsar/pulsar:3.0.4. Moreover, changes are made in the wait strategy to improve the way it handles authentication. A new WaitUntil class has been introduced to manage the authentication token.
---
 src/Testcontainers.Pulsar/PulsarBuilder.cs | 41 ++++++++++++++++------
 1 file changed, 31 insertions(+), 10 deletions(-)

diff --git a/src/Testcontainers.Pulsar/PulsarBuilder.cs b/src/Testcontainers.Pulsar/PulsarBuilder.cs
index 4e84a2fb5..f5445f740 100644
--- a/src/Testcontainers.Pulsar/PulsarBuilder.cs
+++ b/src/Testcontainers.Pulsar/PulsarBuilder.cs
@@ -4,7 +4,7 @@ namespace Testcontainers.Pulsar;
 [PublicAPI]
 public sealed class PulsarBuilder : ContainerBuilder<PulsarBuilder, PulsarContainer, PulsarConfiguration>
 {
-    public const string PulsarImage = "apachepulsar/pulsar:3.0.3";
+    public const string PulsarImage = "apachepulsar/pulsar:3.0.4";
 
     public const ushort PulsarBrokerDataPort = 6650;
 
@@ -84,19 +84,15 @@ 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));
- 
-        //To Do How do I access PulsarContainer.CreateAuthenticationTokenAsync so i can get the auth token?
-        // waitStrategy = waitStrategy.UntilHttpRequestIsSucceeded(request
-        //     => request
-        //         .ForPath("/admin/v2/clusters")
-        //         .ForPort(PulsarWebServicePort)
-        //         .ForResponseMessageMatching(VerifyPulsarStatusAsync)
-        //         .WithHeader("Authorization", ));
+        
+        waitStrategy.AddCustomWaitStrategy(new WaitUntil());
         
         var pulsarBuilder = WithWaitStrategy(waitStrategy);
         
@@ -134,9 +130,34 @@ protected override PulsarBuilder Merge(PulsarConfiguration oldValue, PulsarConfi
         return new PulsarBuilder(new PulsarConfiguration(oldValue, newValue));
     }
     
-    private async Task<bool> VerifyPulsarStatusAsync(System.Net.Http.HttpResponseMessage response)
+    private static async Task<bool> VerifyPulsarStatusAsync(System.Net.Http.HttpResponseMessage response)
     {
         var readAsStringAsync = await response.Content.ReadAsStringAsync();
         return readAsStringAsync == "[\"standalone\"]";
     }
+    
+    /// <inheritdoc cref="IWaitUntil" />
+    private sealed class WaitUntil : IWaitUntil
+    {
+        /// <inheritdoc />
+        public async Task<bool> UntilAsync(IContainer container)
+        {
+            try
+            {
+                var pulsarContainer = container as PulsarContainer;
+                var authenticationToken = await pulsarContainer.CreateAuthenticationTokenAsync(TimeSpan.FromSeconds(60),CancellationToken.None);
+                using (var client = new System.Net.Http.HttpClient())
+                {
+                    client.BaseAddress = new Uri($"http://{container.Hostname}:{container.GetMappedPublicPort(PulsarWebServicePort)}/");
+                    client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", authenticationToken.Replace("\n", ""));
+                    System.Net.Http.HttpResponseMessage response = await client.GetAsync("admin/v2/clusters");
+                    return await VerifyPulsarStatusAsync(response);
+                }
+            }
+            catch (Exception)
+            {
+                return false;
+            }
+        }
+    }
 }
\ No newline at end of file