Make ready for release 2.1.0 and added a sample using BackgroundService and the Process extension method.
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 92783ad..7376c9d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,21 @@
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## [2.1.0] - 2022-02-02
+
+### Added
+
+- Support for on the fly authentication (AuthChallenge)
+- The IAuthentication interface for implementing custom authentication
+- IPulsarClientBuilder.Authentication(IAuthentication authentication) for using custom authentication
+- The AuthenticationFactory with support for:
+ - Token(string token)
+ - Token(Func\<CancellationToken, ValueTask\<string\>\> tokenSupplier)
+
+### Deprecated
+
+- IPulsarClientBuilder.AuthenticateUsingToken(string token) is marked as obsolete. Use Authentication(AuthenticationFactory.Token(...)) instead
+
## [2.0.1] - 2021-11-12
### Changed
diff --git a/DotPulsar.sln b/DotPulsar.sln
index f273d9e..ff10bbf 100644
--- a/DotPulsar.sln
+++ b/DotPulsar.sln
@@ -28,7 +28,9 @@
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotPulsar.IntegrationTests", "tests\DotPulsar.IntegrationTests\DotPulsar.IntegrationTests.csproj", "{B44E52DB-DB45-4E31-AA2C-68E5C52AFDEB}"
EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DotPulsar.TestHelpers", "tests\DotPulsar.TestHelpers\DotPulsar.TestHelpers.csproj", "{F75D1A07-BBED-411C-A900-357790717D8F}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotPulsar.TestHelpers", "tests\DotPulsar.TestHelpers\DotPulsar.TestHelpers.csproj", "{F75D1A07-BBED-411C-A900-357790717D8F}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Processing", "samples\Processing\Processing.csproj", "{CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -68,6 +70,10 @@
{F75D1A07-BBED-411C-A900-357790717D8F}.Debug|Any CPU.Build.0 = Debug|Any CPU
{F75D1A07-BBED-411C-A900-357790717D8F}.Release|Any CPU.ActiveCfg = Release|Any CPU
{F75D1A07-BBED-411C-A900-357790717D8F}.Release|Any CPU.Build.0 = Release|Any CPU
+ {CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -80,6 +86,7 @@
{6D44683B-865C-4D15-9F0A-1A8441354589} = {E7106D0F-B255-4631-9FB8-734FC5748FA9}
{B44E52DB-DB45-4E31-AA2C-68E5C52AFDEB} = {E1C932A9-6D4C-4DDF-8922-BE7B71F12F1C}
{F75D1A07-BBED-411C-A900-357790717D8F} = {E1C932A9-6D4C-4DDF-8922-BE7B71F12F1C}
+ {CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E} = {E7106D0F-B255-4631-9FB8-734FC5748FA9}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {88355922-E70A-4B73-B7F8-ABF8F2B59789}
diff --git a/samples/Consuming/Program.cs b/samples/Consuming/Program.cs
index c16dda1..1ac1b20 100644
--- a/samples/Consuming/Program.cs
+++ b/samples/Consuming/Program.cs
@@ -54,7 +54,7 @@
{
await foreach (var message in consumer.Messages(cancellationToken))
{
- Console.WriteLine("Received: " + message.Value());
+ Console.WriteLine($"Received: {message.Value()}");
await consumer.Acknowledge(message, cancellationToken);
}
}
diff --git a/samples/Processing/Processing.csproj b/samples/Processing/Processing.csproj
new file mode 100644
index 0000000..b5ec451
--- /dev/null
+++ b/samples/Processing/Processing.csproj
@@ -0,0 +1,17 @@
+<Project Sdk="Microsoft.NET.Sdk.Worker">
+
+ <PropertyGroup>
+ <TargetFramework>net6.0</TargetFramework>
+ <Nullable>enable</Nullable>
+ <ImplicitUsings>enable</ImplicitUsings>
+ <UserSecretsId>dotnet-Processing-BFF6CD69-A4C9-4F15-B6AB-A2ED78B89D3B</UserSecretsId>
+ </PropertyGroup>
+
+ <ItemGroup>
+ <PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.0" />
+ </ItemGroup>
+
+ <ItemGroup>
+ <ProjectReference Include="..\..\src\DotPulsar\DotPulsar.csproj" />
+ </ItemGroup>
+</Project>
diff --git a/samples/Processing/Program.cs b/samples/Processing/Program.cs
new file mode 100644
index 0000000..ddb38c3
--- /dev/null
+++ b/samples/Processing/Program.cs
@@ -0,0 +1,21 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using Processing;
+
+await Host
+ .CreateDefaultBuilder(args)
+ .ConfigureServices(services => services.AddHostedService<Worker>())
+ .Build()
+ .RunAsync();
diff --git a/samples/Processing/Properties/launchSettings.json b/samples/Processing/Properties/launchSettings.json
new file mode 100644
index 0000000..2906cde
--- /dev/null
+++ b/samples/Processing/Properties/launchSettings.json
@@ -0,0 +1,11 @@
+{
+ "profiles": {
+ "Processing": {
+ "commandName": "Project",
+ "dotnetRunMessages": true,
+ "environmentVariables": {
+ "DOTNET_ENVIRONMENT": "Development"
+ }
+ }
+ }
+}
diff --git a/samples/Processing/Worker.cs b/samples/Processing/Worker.cs
new file mode 100644
index 0000000..d4d639d
--- /dev/null
+++ b/samples/Processing/Worker.cs
@@ -0,0 +1,62 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace Processing;
+
+using DotPulsar;
+using DotPulsar.Abstractions;
+using DotPulsar.Extensions;
+
+public class Worker : BackgroundService
+{
+ private readonly ILogger<Worker> _logger;
+
+ public Worker(ILogger<Worker> logger) => _logger = logger;
+
+ protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+ {
+ await using var client = PulsarClient.Builder().Build(); //Connecting to pulsar://localhost:6650
+
+ await using var consumer = client.NewConsumer(Schema.String)
+ .StateChangedHandler(Monitor, cancellationToken)
+ .SubscriptionName("MySubscription")
+ .Topic("persistent://public/default/mytopic")
+ .Create();
+
+ await consumer.Process(ProcessMessage, cancellationToken);
+ }
+
+ private ValueTask ProcessMessage(IMessage<string> message, CancellationToken cancellationToken)
+ {
+ _logger.LogInformation($"Received: {message.Value()}");
+ return ValueTask.CompletedTask;
+ }
+
+ private void Monitor(ConsumerStateChanged stateChanged, CancellationToken cancellationToken)
+ {
+ var stateMessage = stateChanged.ConsumerState switch
+ {
+ ConsumerState.Active => "is active",
+ ConsumerState.Inactive => "is inactive",
+ ConsumerState.Disconnected => "is disconnected",
+ ConsumerState.Closed => "has closed",
+ ConsumerState.ReachedEndOfTopic => "has reached end of topic",
+ ConsumerState.Faulted => "has faulted",
+ _ => $"has an unknown state '{stateChanged.ConsumerState}'"
+ };
+
+ var topic = stateChanged.Consumer.Topic;
+ _logger.LogInformation($"The consumer for topic '{topic}' {stateMessage}");
+ }
+}
diff --git a/samples/Processing/appsettings.Development.json b/samples/Processing/appsettings.Development.json
new file mode 100644
index 0000000..b2dcdb6
--- /dev/null
+++ b/samples/Processing/appsettings.Development.json
@@ -0,0 +1,8 @@
+{
+ "Logging": {
+ "LogLevel": {
+ "Default": "Information",
+ "Microsoft.Hosting.Lifetime": "Information"
+ }
+ }
+}
diff --git a/samples/Processing/appsettings.json b/samples/Processing/appsettings.json
new file mode 100644
index 0000000..b2dcdb6
--- /dev/null
+++ b/samples/Processing/appsettings.json
@@ -0,0 +1,8 @@
+{
+ "Logging": {
+ "LogLevel": {
+ "Default": "Information",
+ "Microsoft.Hosting.Lifetime": "Information"
+ }
+ }
+}
diff --git a/samples/Producing/Program.cs b/samples/Producing/Program.cs
index 0dce9f5..79045a3 100644
--- a/samples/Producing/Program.cs
+++ b/samples/Producing/Program.cs
@@ -57,7 +57,7 @@
{
var data = DateTime.UtcNow.ToLongTimeString();
_ = await producer.Send(data, cancellationToken);
- Console.WriteLine("Sent: " + data);
+ Console.WriteLine($"Sent: {data}");
await Task.Delay(delay, cancellationToken);
}
}
diff --git a/samples/Reading/Program.cs b/samples/Reading/Program.cs
index d00ad5c..88b9f7c 100644
--- a/samples/Reading/Program.cs
+++ b/samples/Reading/Program.cs
@@ -54,7 +54,7 @@
{
await foreach (var message in reader.Messages(cancellationToken))
{
- Console.WriteLine("Received: " + message.Value());
+ Console.WriteLine($"Received: {message.Value()}");
}
}
catch (OperationCanceledException) { }
diff --git a/src/DotPulsar/AuthenticationFactory.cs b/src/DotPulsar/AuthenticationFactory.cs
index 6ed3dfb..2fadc67 100644
--- a/src/DotPulsar/AuthenticationFactory.cs
+++ b/src/DotPulsar/AuthenticationFactory.cs
@@ -21,7 +21,7 @@
using System.Threading.Tasks;
/// <summary>
-/// Factory class that allows to create Authentication instances for all the supported authentication methods.
+/// Factory for creating IAuthentication instances for all the supported authentication methods.
/// </summary>
public static class AuthenticationFactory
{
diff --git a/src/DotPulsar/DotPulsar.csproj b/src/DotPulsar/DotPulsar.csproj
index cf976bd..56a4b92 100644
--- a/src/DotPulsar/DotPulsar.csproj
+++ b/src/DotPulsar/DotPulsar.csproj
@@ -2,7 +2,7 @@
<PropertyGroup>
<TargetFrameworks>netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0</TargetFrameworks>
- <Version>2.0.1</Version>
+ <Version>2.1.0</Version>
<AssemblyVersion>$(Version)</AssemblyVersion>
<FileVersion>$(Version)</FileVersion>
<Authors>ApachePulsar,DanskeCommodities,dblank</Authors>