Support http lookup authentication(oauth2 and token)

Author: ninjazhou

src/Pulsar.Client/Api/AuthenticationDataProvider.fs

@@ -1,5 +1,6 @@
 namespace Pulsar.Client.Api
 
+open System.Collections.Generic
 open Pulsar.Client.Common
 open System.Text
 open System.Security.Cryptography.X509Certificates
@@ -26,6 +27,18 @@ type AuthenticationDataProvider() =
     default this.GetCommandData() =
         ""
 
+    //Http
+
+    abstract member HasDataForHttp: unit -> bool
+    default this.HasDataForHttp() =
+        false
+
+    abstract member GetHttpHeaders: unit -> IReadOnlyDictionary<string, string>
+    default this.GetHttpHeaders() =
+        Dictionary<string, string>()
+
+    //AuthDataProvider
+
     abstract member Authenticate: AuthData -> AuthData
     default this.Authenticate authData =
         let bytes =

src/Pulsar.Client/Auth/AuthenticationDataToken.fs

@@ -1,5 +1,6 @@
 namespace Pulsar.Client.Auth    
 
+open System.Collections.Generic
 open Pulsar.Client.Api
 
 type internal AuthenticationDataToken (supplier: unit -> string) =
@@ -9,4 +10,18 @@ type internal AuthenticationDataToken (supplier: unit -> string) =
         true
 
     override this.GetCommandData() =
-        supplier()
+        supplier()
+
+    override this.HasDataForHttp()=
+        true
+
+    //  Since AuthenticationOauth2 and AuthenticationToken both return this class when call GetAuthData()
+    //  We only need to realize this DataToken class GetHttpHeaders to realize http authentication
+    override this.GetHttpHeaders()=
+        let httpHeaderPropertiesDict = Dictionary<string, string>()
+        httpHeaderPropertiesDict.Add("X-Pulsar-Auth-Method-Name", "token")
+        httpHeaderPropertiesDict.Add("Authorization", "Bearer " + supplier())
+        httpHeaderPropertiesDict
+
+
+

src/Pulsar.Client/Internal/HttpLookupService.fs

@@ -15,10 +15,8 @@ open Pulsar.Client.Schema
 
 type internal HttpLookupService (config: PulsarClientConfiguration, _connectionPool: ConnectionPool) =
 
-    let httpClient = new HttpClient(new SocketsHttpHandler(
-        PooledConnectionLifetime = TimeSpan.FromMinutes(2),
-        AllowAutoRedirect = true
-    ))
+    let pulsarHttpClient = PulsarHttpClient(config)
+
     let jsonOptions = JsonSerializerOptions(
         PropertyNamingPolicy = JsonNamingPolicy.CamelCase
      )
@@ -93,7 +91,7 @@ type internal HttpLookupService (config: PulsarClientConfiguration, _connectionP
                 let topicRestPath = topic.Replace("persistent://","persistent/").Replace("non-persistent://","non-persistent/")
                 let! response =
                     randomServiceUri.AbsoluteUri + $"admin/v2/%s{topicRestPath}/partitions?checkAllowAutoCreation=true"
-                    |> httpClient.GetStreamAsync
+                    |> pulsarHttpClient.GetStreamAsync
                     |> Async.AwaitTask
                 let brokerResponse = JsonSerializer.Deserialize<{| Partitions: int |}>(response, jsonOptions)
                 return { Partitions = brokerResponse.Partitions }
@@ -113,7 +111,7 @@ type internal HttpLookupService (config: PulsarClientConfiguration, _connectionP
             let randomServiceUri = config.ServiceAddresses[RandomGenerator.Next(0, config.ServiceAddresses.Length)]
             let topic: string = %topicName
             let topicRestPath = topic.Replace("persistent://","persistent/").Replace("non-persistent://","non-persistent/")
-            let! response = httpClient.GetStreamAsync (randomServiceUri.AbsoluteUri + $"lookup/v2/topic/%s{topicRestPath}")
+            let! response = pulsarHttpClient.GetStreamAsync (randomServiceUri.AbsoluteUri + $"lookup/v2/topic/%s{topicRestPath}")
             let brokerResponse =
                 JsonSerializer.Deserialize<{|
                     BrokerUrl : string
@@ -142,7 +140,7 @@ type internal HttpLookupService (config: PulsarClientConfiguration, _connectionP
                     | false -> "NON_PERSISTENT"
                 let! response =
                     randomServiceUri.AbsoluteUri + $"admin/v2/namespaces/%s{ns.ToString()}/topics?mode=%s{mode}"
-                    |> httpClient.GetStreamAsync
+                    |> pulsarHttpClient.GetStreamAsync
                     |> Async.AwaitTask
                 let brokerResponse = JsonSerializer.Deserialize<string seq>(response, jsonOptions)
                 return brokerResponse
@@ -176,7 +174,7 @@ type internal HttpLookupService (config: PulsarClientConfiguration, _connectionP
                         $"admin/v2/schemas/%s{topicRestPath}/schema"
                 let! response =
                     randomServiceUri.AbsoluteUri + path
-                    |> httpClient.GetStreamAsync
+                    |> pulsarHttpClient.GetStreamAsync
                     |> Async.AwaitTask
                 let schemaResponse =
                     JsonSerializer.Deserialize<{|

src/Pulsar.Client/Internal/PulsarHttpClient.fs

@@ -0,0 +1,35 @@
+namespace Pulsar.Client.Internal
+
+open System.Net.Http
+open Pulsar.Client.Api
+open System
+
+//  This class is mainly used for http lookup service
+//  We name this class `PulsarHttpClient` to avoid naming clash with native HttpClient,
+//  and in Java it's just `HttpClient`
+type internal PulsarHttpClient (config: PulsarClientConfiguration) =
+
+    let authenticationDataProvider = config.Authentication.GetAuthData()
+
+    let httpClient = new HttpClient(new SocketsHttpHandler(
+        PooledConnectionLifetime = TimeSpan.FromMinutes(2),
+        AllowAutoRedirect = true
+    ))
+
+    member this.GetStreamAsync (requestUri: string) =
+        backgroundTask {
+            if authenticationDataProvider.HasDataForHttp() then
+                let request = new HttpRequestMessage(HttpMethod.Get, requestUri)
+                for headerPropertyEntry in authenticationDataProvider.GetHttpHeaders() do
+                    request.Headers.Add(headerPropertyEntry.Key, headerPropertyEntry.Value)
+                let! response = httpClient.SendAsync(request, HttpCompletionOption.ResponseContentRead)
+                response.EnsureSuccessStatusCode() |> ignore
+                return! response.Content.ReadAsStreamAsync()
+            else
+                return! httpClient.GetStreamAsync(requestUri)
+        }
+
+
+
+
+

src/Pulsar.Client/Pulsar.Client.fsproj

@@ -139,6 +139,7 @@
     <Compile Include="Internal\EndPointResolver.fs" />
     <Compile Include="Internal\ILookupService.fs" />
     <Compile Include="Internal\BinaryLookupService.fs" />
+    <Compile Include="Internal\PulsarHttpClient.fs" />
     <Compile Include="Internal\HttpLookupService.fs" />
     <Compile Include="Internal\ConnectionHandler.fs" />
     <Compile Include="Internal\AcknowledgmentsGroupingTracker.fs" />

tests/UnitTests/Api/PulsarClientBuilderTests.fs

@@ -39,4 +39,17 @@ module PulsarClientBuilderTests =
                     "Service Url needs to be specified on the PulsarClientBuilder object."
             }
 
+            test "Http lookup authentication authDataProvider" {
+                AuthenticationFactory.Token("test").GetAuthData().HasDataForHttp()
+                |> Expect.equal "AuthenticationToken HasDataForHttp should be true" true
+
+                AuthenticationFactoryOAuth2.ClientCredentials(
+                    Uri("https://test.com"),
+                    "test",
+                    Uri("https://test.com")
+                ).GetAuthMethodName()
+                |> Expect.equal "AuthenticationFactoryOAuth2 authData should be token" "token"
+            }
+
+
         ]