This is an automated email from the ASF dual-hosted git repository.
blankensteiner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-dotpulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 22ff109 Make ready for 0.11.0
22ff109 is described below
commit 22ff109bf157254237fb598bd1efeaeec2b0564a
Author: Daniel Blankensteiner <[email protected]>
AuthorDate: Sun Feb 21 11:38:24 2021 +0100
Make ready for 0.11.0
---
CHANGELOG.md | 48 ++++++++++++++-
README.md | 42 +++----------
src/DotPulsar/DotPulsar.csproj | 13 +---
src/DotPulsar/Extensions/ConsumerExtensions.cs | 74 -----------------------
src/DotPulsar/Internal/DotPulsarActivitySource.cs | 28 ---------
5 files changed, 57 insertions(+), 148 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index aaf2c1e..0081ed9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,50 @@ All notable changes to this project will be documented in this
file.
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).
+## [0.11.0] - 2021-02-21
+
+### Added
+
+- The Consumer and Reader now share the IReceive interface for receiving a
single message: ValueTask\<Message\> Receive(CancellationToken
cancellationToken)
+- The Consumer and Reader now share the ISeek interface for seeking on
message-id and publish time
+- The Consumer and Reader now share the IGetLastMessageId interface for
getting the last message-id on a topic
+- The Consumer, Reader, and Producer now share the IState interface adding
'OnStateChangeFrom' and 'OnStateChangeTo'
+- The PulsarClient, Consumer, Reader, and Producer now have the read-only
property 'ServiceUrl'
+- The Consumer now have the read-only property 'SubscriptionName'
+- All message compression types are now supported (listed below). Please read
this [compression
how-to](https://github.com/apache/pulsar-dotpulsar/wiki/Compression)
+ - LZ4
+ - SNAPPY
+ - ZLIB
+ - ZSTD
+
+### Changed
+
+- MessageId.ToString() now returns a string matching that of other clients:
"{LedgerId}:{EntryId}:{Partition}:{BatchIndex}"
+- A lot of methods were made into extension methods and now require a using
statement for 'DotPulsar.Extensions'
+ - Producer.StateChangedTo(ProducerState state, CancellationToken
cancellationToken)
+ - Producer.StateChangedFrom(ProducerState state, CancellationToken
cancellationToken)
+ - Producer.Send(byte[] data, CancellationToken cancellationToken)
+ - Producer.Send(ReadOnlyMemory\<byte\> data, CancellationToken
cancellationToken)
+ - Producer.Send(MessageMetadata metadata, byte[] data,
CancellationToken cancellationToken)
+ - Producer.Send(MessageMetadata metadata, ReadOnlyMemory\<byte\> data,
CancellationToken cancellationToken)
+ - Consumer.Acknowledge(Message message, CancellationToken
cancellationToken)
+ - Consumer.AcknowledgeCumulative(Message message, CancellationToken
cancellationToken)
+ - Consumer.StateChangedTo(ConsumerState state, CancellationToken
cancellationToken)
+ - Consumer.StateChangedFrom(ConsumerState state, CancellationToken
cancellationToken)
+ - Consumer.Messages(CancellationToken cancellationToken)
+ - Consumer.Seek(DateTime publishTime, CancellationToken
cancellationToken)
+ - Consumer.Seek(DateTimeOffset publishTime, CancellationToken
cancellationToken)
+ - Reader.StateChangedTo(ReaderState state, CancellationToken
cancellationToken)
+ - Reader.StateChangedFrom(ReaderState state, CancellationToken
cancellationToken)
+ - Reader.Messages(CancellationToken cancellationToken)
+ - Reader.Seek(DateTime publishTime, CancellationToken cancellationToken)
+ - Reader.Seek(DateTimeOffset publishTime, CancellationToken
cancellationToken)
+
+### Fixed
+
+- Before the Consumer and Reader would throw an ArgumentOutOfRangeException if
they encountered a compressed message. Now they will throw a
CompressionException if the compression type is not supported
+- DotPulsarEventSource (performance counters) was only enabled for .NET
Standard 2.1. Now it's enabled for all target frameworks except .NET Standard
2.0
+
## [0.10.1] - 2020-12-23
### Added
@@ -12,7 +56,7 @@ The format is based on [Keep a
Changelog](https://keepachangelog.com/en/1.0.0/),
### Fixed
-- Do not throw exceptions when disposing consumers, readers or producers
+- Do not throw exceptions when disposing consumers, readers, or producers
## [0.10.0] - 2020-12-16
@@ -20,7 +64,7 @@ The format is based on [Keep a
Changelog](https://keepachangelog.com/en/1.0.0/),
- Added performance improvements when receiving data
- Added the IHandleStateChanged\<TStateChanged\> interface for easier state
monitoring
-- Added StateChangedHandler methods on ConsumerBuilder, ReaderBuilder and
ProducerBuilder for easier state monitoring
+- Added StateChangedHandler methods on ConsumerBuilder, ReaderBuilder, and
ProducerBuilder for easier state monitoring
- Added StateChangedHandler property on ConsumerOptions, ReaderOptions, and
ProducerOptions for easier state monitoring
- Added four new DotPulsarExceptions: InvalidTransactionStatusException,
NotAllowedException, TransactionConflictException and
TransactionCoordinatorNotFoundException
- Added properties on Message to read EventTime and PublishTime as DateTime
diff --git a/README.md b/README.md
index c42e740..fe7afdc 100644
--- a/README.md
+++ b/README.md
@@ -2,11 +2,9 @@
-.NET/C# client library for [Apache Pulsar](https://pulsar.apache.org/).
+The official .NET/C# client library for [Apache
Pulsar](https://pulsar.apache.org/).
-DotPulsar is written entirely in C# and implements Apache Pulsar's [binary
protocol](https://pulsar.apache.org/docs/en/develop-binary-protocol/). Other
options was using the [C++ client
library](https://pulsar.apache.org/docs/en/client-libraries-cpp/) (which is
what the [Python
client](https://pulsar.apache.org/docs/en/client-libraries-python/) and [Go
client](https://pulsar.apache.org/docs/en/client-libraries-go/) do) or build on
top of the [WebSocket API](https://pulsar.apache.org/docs [...]
-
-DotPulsar's API is strongly inspired by Apache Pulsar's official [Java
client](https://pulsar.apache.org/docs/en/client-libraries-java/), but a 100%
match is not a goal.
+DotPulsar is written entirely in C# and implements Apache Pulsar's [binary
protocol](https://pulsar.apache.org/docs/en/develop-binary-protocol/).
## What's new?
@@ -52,48 +50,26 @@ For a more in-depth tour of the API, please visit the
[Wiki](https://github.com/
- [X] TLS Authentication
- [X] JSON Web Token Authentication
- [X] Producer send with custom metadata
-- [X] Producer send with event time, sequence id and delayed message delivery
+- [X] Producer send with event time, sequence id, and delayed message delivery
- [X] Producer send with key and ordering key
- [X] Consumer subscription with initial position and priority level
-- [X] Consumer subscription types exclusive, shared, failover and key shared
+- [X] Consumer subscription types exclusive, shared, failover, and key shared
- [X] Consumer receive and single + cumulative acknowledge
-- [X] Consumer and Reader seek on message id and publish time
+- [X] Consumer and Reader seek on message-id and publish time
- [X] Consumer unsubscribe
- [X] Consume compacted topics
- [X] Reader API
- [X] Read/Consume/Acknowledge batched messages
- [X] Pulsar Proxy
+- [X] [LZ4 message
compression](https://github.com/apache/pulsar-dotpulsar/wiki/Compression)
+- [X] [ZLIB message
compression](https://github.com/apache/pulsar-dotpulsar/wiki/Compression)
+- [X] [ZSTD message
compression](https://github.com/apache/pulsar-dotpulsar/wiki/Compression)
+- [X] [SNAPPY message
compression](https://github.com/apache/pulsar-dotpulsar/wiki/Compression)
## Roadmap
Help prioritizing the roadmap is most welcome, so please reach out and tell us
what you want and need.
-### 1.0.0
-
-We are feature complete for this release. We just need testing.
-
-- [X] Use IAsyncDisposable
-- [X] Use IAsyncEnumerable
-- [X] Use ValueTask instead of Task
-- [X] Make solution nullable
-- [X] Support .NET Standard 2.0 and 2.1
-
-### If requested by the community
-
-Let us know which features you need by creating an issue or by giving existing
issues a "Thumbs up".
-
-* [Message encryption](https://github.com/apache/pulsar-dotpulsar/issues/8)
-* [Batching when
producing](https://github.com/apache/pulsar-dotpulsar/issues/7)
-* [Schema](https://github.com/apache/pulsar-dotpulsar/issues/6)
-* [Partitioned topics](https://github.com/apache/pulsar-dotpulsar/issues/4)
-* [Multi-topic
subscriptions](https://github.com/apache/pulsar-dotpulsar/issues/5)
-* Athenz Authentication
-* Kerberos Authentication
-* LZ4 message compression
-* ZLIB message compression
-* ZSTD message compression
-* SNAPPY message compression
-
## Join Our Community
Apache Pulsar has a [Slack instance](https://pulsar.apache.org/contact/) and
there you'll find us in the #dev-dotpulsar channel. Just waiting for you to pop
by :-)
diff --git a/src/DotPulsar/DotPulsar.csproj b/src/DotPulsar/DotPulsar.csproj
index b22d835..f89de85 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</TargetFrameworks>
- <Version>0.10.1</Version>
+ <Version>0.11.0</Version>
<AssemblyVersion>$(Version)</AssemblyVersion>
<FileVersion>$(Version)</FileVersion>
<Authors>DanskeCommodities;dblank</Authors>
@@ -12,7 +12,7 @@
<PackageTags>Apache;Pulsar</PackageTags>
<PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
<PackageReleaseNotes>Please refer to CHANGELOG.md for
details</PackageReleaseNotes>
- <Description>.NET/C# client library for Apache Pulsar</Description>
+ <Description>The official .NET/C# client library for Apache
Pulsar</Description>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<IncludeSymbols>true</IncludeSymbols>
@@ -30,15 +30,6 @@
<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
<PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="5.0.0"
/>
<PackageReference Include="Microsoft.Bcl.HashCode" Version="1.1.1" />
- <PackageReference Include="System.Diagnostics.DiagnosticSource"
Version="5.0.1" />
- </ItemGroup>
-
- <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.1'">
- <PackageReference Include="System.Diagnostics.DiagnosticSource"
Version="5.0.1" />
- </ItemGroup>
-
- <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1'">
- <PackageReference Include="System.Diagnostics.DiagnosticSource"
Version="5.0.1" />
</ItemGroup>
</Project>
diff --git a/src/DotPulsar/Extensions/ConsumerExtensions.cs
b/src/DotPulsar/Extensions/ConsumerExtensions.cs
index 63b8d49..b61971c 100644
--- a/src/DotPulsar/Extensions/ConsumerExtensions.cs
+++ b/src/DotPulsar/Extensions/ConsumerExtensions.cs
@@ -15,10 +15,6 @@
namespace DotPulsar.Extensions
{
using DotPulsar.Abstractions;
- using DotPulsar.Internal;
- using System;
- using System.Collections.Generic;
- using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
@@ -40,76 +36,6 @@ namespace DotPulsar.Extensions
=> await consumer.AcknowledgeCumulative(message.MessageId,
cancellationToken).ConfigureAwait(false);
/// <summary>
- /// Process and auto-acknowledge a message.
- /// </summary>
- public static async ValueTask Process(this IConsumer consumer,
Func<Message, CancellationToken, ValueTask> processor, CancellationToken
cancellationToken = default) // TODO Allow user to set number of workers
- {
- const string operation = "process";
- var operationName = $"{consumer.Topic} {operation}";
-
- var tags = new List<KeyValuePair<string, object?>>
- {
- new KeyValuePair<string, object?>("messaging.destination",
consumer.Topic),
- new KeyValuePair<string,
object?>("messaging.destination_kind", "topic"),
- new KeyValuePair<string, object?>("messaging.operation",
operation),
- new KeyValuePair<string, object?>("messaging.system",
"pulsar"),
- new KeyValuePair<string, object?>("messaging.url",
consumer.ServiceUrl),
- new KeyValuePair<string,
object?>("messaging.pulsar.subscription", consumer.SubscriptionName) // TODO
Ask Pulsar community to define Pulsar specific tags
- };
-
- while (!cancellationToken.IsCancellationRequested)
- {
- var message = await consumer.Receive(cancellationToken);
-
- var activity = StartActivity(message, operationName, tags);
-
- if (activity is not null && activity.IsAllDataRequested)
- {
- activity.SetTag("messaging.message_id",
message.MessageId.ToString());
- activity.SetTag("messaging.message_payload_size_bytes",
message.Data.Length);
- }
-
- try
- {
- await processor(message, cancellationToken);
- }
- catch
- {
- // Ignore
- }
-
- activity?.Dispose();
-
- await consumer.Acknowledge(message.MessageId,
cancellationToken);
- }
- }
-
- private static Activity? StartActivity(Message message, string
operationName, IEnumerable<KeyValuePair<string, object?>> tags)
- {
- if (!DotPulsarActivitySource.ActivitySource.HasListeners())
- return null;
-
- var properties = message.Properties;
-
- if (properties.TryGetValue("traceparent", out var traceparent))
// TODO Allow the user to overwrite the keys 'traceparent' and 'tracestate'
- {
- var tracestate = properties.ContainsKey("tracestate") ?
properties["tracestrate"] : null;
- if (ActivityContext.TryParse(traceparent, tracestate, out var
activityContext))
- return
DotPulsarActivitySource.ActivitySource.StartActivity(operationName,
ActivityKind.Consumer, activityContext, tags);
- }
-
- var activity =
DotPulsarActivitySource.ActivitySource.StartActivity(operationName,
ActivityKind.Consumer);
-
- if (activity is not null && activity.IsAllDataRequested)
- {
- foreach (var tag in tags)
- activity.SetTag(tag.Key, tag.Value);
- }
-
- return activity;
- }
-
- /// <summary>
/// Wait for the state to change to a specific state.
/// </summary>
/// <returns>
diff --git a/src/DotPulsar/Internal/DotPulsarActivitySource.cs
b/src/DotPulsar/Internal/DotPulsarActivitySource.cs
deleted file mode 100644
index 09c8464..0000000
--- a/src/DotPulsar/Internal/DotPulsarActivitySource.cs
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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 DotPulsar.Internal
-{
- using System.Diagnostics;
-
- public static class DotPulsarActivitySource
- {
- static DotPulsarActivitySource()
- {
- ActivitySource = new ActivitySource(Constants.ClientName,
Constants.ClientVersion);
- }
-
- public static ActivitySource ActivitySource { get; }
- }
-}
