You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bl...@apache.org on 2023/01/30 11:09:58 UTC
[pulsar-dotpulsar] branch master updated: Fix tests and have 'Process' also throw an error when faulting
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 15214dd Fix tests and have 'Process' also throw an error when faulting
15214dd is described below
commit 15214dde6df69375f7bf69c2638b1aa8106629e1
Author: Daniel Blankensteiner <db...@vmail.dk>
AuthorDate: Mon Jan 30 12:09:50 2023 +0100
Fix tests and have 'Process' also throw an error when faulting
---
CHANGELOG.md | 10 ++++++++++
samples/Processing/Processing.csproj | 2 +-
src/DotPulsar/Internal/MessageProcessor.cs | 8 ++++++--
src/DotPulsar/Internal/Producer.cs | 4 ++++
tests/DotPulsar.Tests/TokenTests.cs | 5 +++--
5 files changed, 24 insertions(+), 5 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 359159f..b945fe1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,16 @@ 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).
+## [Unreleased]
+
+### Added
+
+- ConsumerFaultedException, ProducerFaultedException, ReaderFaultedException that all inherit from FaultedException. The inner exception will show why it faulted
+
+### Changed
+
+- If a consumer, reader, or producer is faulted all method calls will throw a ConsumerFaultedException, ProducerFaultedException, or ReaderFaultedException
+
## [2.9.0] - 2023-01-26
### Added
diff --git a/samples/Processing/Processing.csproj b/samples/Processing/Processing.csproj
index 4565e5f..943908b 100644
--- a/samples/Processing/Processing.csproj
+++ b/samples/Processing/Processing.csproj
@@ -4,7 +4,7 @@
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
- <UserSecretsId>dotnet-Processing-BFF6CD69-A4C9-4F15-B6AB-A2ED78B89D3B</UserSecretsId>
+ <NoWarn>CA2012</NoWarn>
</PropertyGroup>
<ItemGroup>
diff --git a/src/DotPulsar/Internal/MessageProcessor.cs b/src/DotPulsar/Internal/MessageProcessor.cs
index d2bed80..5d5c32e 100644
--- a/src/DotPulsar/Internal/MessageProcessor.cs
+++ b/src/DotPulsar/Internal/MessageProcessor.cs
@@ -1,4 +1,4 @@
-/*
+/*
* 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
@@ -21,6 +21,7 @@ using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
+using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
@@ -91,11 +92,14 @@ public sealed class MessageProcessor<TMessage> : IDisposable
StartNewProcessorTask(cancellationToken);
}
- while (!cancellationToken.IsCancellationRequested && !_consumer.IsFinalState())
+ while (true)
{
StartNewProcessorTask(cancellationToken);
var completedTask = await Task.WhenAny(_processorTasks).ConfigureAwait(false);
+ if (completedTask.IsFaulted)
+ ExceptionDispatchInfo.Capture(completedTask.Exception!.InnerException!).Throw();
_processorTasks.Remove(completedTask);
+ cancellationToken.ThrowIfCancellationRequested();
}
}
diff --git a/src/DotPulsar/Internal/Producer.cs b/src/DotPulsar/Internal/Producer.cs
index c9d608a..be92548 100644
--- a/src/DotPulsar/Internal/Producer.cs
+++ b/src/DotPulsar/Internal/Producer.cs
@@ -248,7 +248,11 @@ public sealed class Producer<TMessage> : IProducer<TMessage>, IRegisterEvent
ValueTask OnMessageSent(MessageId messageId)
{
tcs.TrySetResult(messageId);
+#if NET6_0_OR_GREATER
+ return ValueTask.CompletedTask;
+#else
return new ValueTask();
+#endif
}
await InternalSend(metadata, message, true, OnMessageSent, cancellationToken).ConfigureAwait(false);
diff --git a/tests/DotPulsar.Tests/TokenTests.cs b/tests/DotPulsar.Tests/TokenTests.cs
index 9734fd9..3759125 100644
--- a/tests/DotPulsar.Tests/TokenTests.cs
+++ b/tests/DotPulsar.Tests/TokenTests.cs
@@ -50,7 +50,7 @@ public class TokenTests
var state = await producer.OnStateChangeTo(ProducerState.Faulted);
// Assert
- exception.Should().BeOfType<Exception>();
+ exception.Should().BeOfType<ProducerFaultedException>();
state.Should().Be(ProducerState.Faulted);
}
@@ -96,7 +96,8 @@ public class TokenTests
var state = await producer.OnStateChangeTo(ProducerState.Faulted);
// Assert
- exception.Should().BeOfType<AuthenticationException>();
+ exception.Should().BeOfType<ProducerFaultedException>();
+ exception.InnerException.Should().BeOfType<AuthenticationException>();
state.Should().Be(ProducerState.Faulted);
}