This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton-dotnet.git
The following commit(s) were added to refs/heads/main by this push:
new 8044c68 PROTON-2915 Improve the delivery state APIs for easier use
8044c68 is described below
commit 8044c68d4989b260e8b38b608267b69212c67423
Author: Timothy Bish <[email protected]>
AuthorDate: Wed Jan 14 14:36:57 2026 -0500
PROTON-2915 Improve the delivery state APIs for easier use
Make some improvements the APIs around Delivery and Tracker types
DeliveryState
types for local and remote states such that users can get easier access to
the
actual data in a DeliveryState like Modified or Rejected where information
might be useful to diagnose the outcomes.
---
src/Proton.Client/Client/IAccepted.cs | 28 ++++
src/Proton.Client/Client/IDeliveryState.cs | 30 +++-
src/Proton.Client/Client/IModified.cs | 44 +++++
src/Proton.Client/Client/IRejected.cs | 44 +++++
src/Proton.Client/Client/IReleased.cs | 28 ++++
src/Proton.Client/Client/ITransactional.cs | 32 ++++
.../Client/Implementation/ClientDeliveryState.cs | 84 +++++++++-
src/Proton/Types/Transport/ErrorCondition.cs | 3 +-
.../Implementation/ClientDeliveryStateTest.cs | 181 +++++++++++++++++++++
.../Client/Implementation/ClientReceiverTest.cs | 59 +++++++
.../Client/Implementation/ClientSenderTest.cs | 58 +++++++
.../Implementation/ClilentTransactionsTest.cs | 82 ++++++++++
12 files changed, 661 insertions(+), 12 deletions(-)
diff --git a/src/Proton.Client/Client/IAccepted.cs
b/src/Proton.Client/Client/IAccepted.cs
new file mode 100644
index 0000000..f08a34b
--- /dev/null
+++ b/src/Proton.Client/Client/IAccepted.cs
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 Apache.Qpid.Proton.Client.Implementation;
+
+namespace Apache.Qpid.Proton.Client
+{
+ /// <summary>
+ /// Marker interface for Accepted delivery states and outcomes
+ /// </summary>
+ public interface IAccepted : IDeliveryState
+ {
+ }
+}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/IDeliveryState.cs
b/src/Proton.Client/Client/IDeliveryState.cs
index 4c23cab..0d2c909 100644
--- a/src/Proton.Client/Client/IDeliveryState.cs
+++ b/src/Proton.Client/Client/IDeliveryState.cs
@@ -15,10 +15,16 @@
* limitations under the License.
*/
+using System.Collections.Generic;
using Apache.Qpid.Proton.Client.Implementation;
namespace Apache.Qpid.Proton.Client
{
+ /// <summary>
+ /// Delivery state instance have a type (e.g. Accepted, Rejected ...) and
can in
+ /// case of a transactional delivery state, carry an outcome and in those
cases the is
+ /// outcome APIs return the value of the nested outcome carried in the
transactional
+ /// </summary>
public interface IDeliveryState
{
/// <summary>
@@ -32,6 +38,26 @@ namespace Apache.Qpid.Proton.Client
/// </summary>
bool IsAccepted { get; }
+ /// <summary>
+ /// Quick access to determine if the state value indicates the delivery
was rejected.
+ /// </summary>
+ bool IsRejected { get; }
+
+ /// <summary>
+ /// Quick access to determine if the state value indicates the delivery
was released.
+ /// </summary>
+ bool IsReleased { get; }
+
+ /// <summary>
+ /// Quick access to determine if the state value indicates the delivery
was modified.
+ /// </summary>
+ bool IsModified { get; }
+
+ /// <summary>
+ /// Quick access to determine if the state value indicates the delivery
was transactional.
+ /// </summary>
+ bool IsTransactional { get; }
+
/// <summary>
/// Returns an instance of a delivery state that accepts a delivery
/// </summary>
@@ -48,13 +74,13 @@ namespace Apache.Qpid.Proton.Client
/// Returns an instance of a delivery state that rejects a delivery
/// </summary>
/// <returns>An rejected delivery state type</returns>
- static IDeliveryState Rejected(string condition, string description =
null) => new ClientRejected(condition, description);
+ static IDeliveryState Rejected(string condition, string description =
null, IDictionary<string, object> info = null) => new ClientRejected(condition,
description, info);
/// <summary>
/// Returns an instance of a delivery state that modifies a delivery
/// </summary>
/// <returns>An modified delivery state type</returns>
- static IDeliveryState Modified(bool deliveryFailed, bool
undeliverableHere = false) => new ClientModified(deliveryFailed,
undeliverableHere);
+ static IDeliveryState Modified(bool deliveryFailed, bool
undeliverableHere = false, IDictionary<string, object> annotations = null) =>
new ClientModified(deliveryFailed, undeliverableHere, annotations);
}
}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/IModified.cs
b/src/Proton.Client/Client/IModified.cs
new file mode 100644
index 0000000..dca0c0e
--- /dev/null
+++ b/src/Proton.Client/Client/IModified.cs
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 System.Collections.Generic;
+using Apache.Qpid.Proton.Client.Implementation;
+
+namespace Apache.Qpid.Proton.Client
+{
+ /// <summary>
+ /// Marker interface for Modified delivery states and outcomes
+ /// </summary>
+ public interface IModified : IDeliveryState
+ {
+ /// <summary>
+ /// Quick access to determine if the modified outcome indicates the
delivery has failed.
+ /// </summary>
+ bool DeliveryFailed { get; }
+
+ /// <summary>
+ /// Quick access to determine if the modified outcome indicates the
delivery cannot be redelivered to the target.
+ /// </summary>
+ bool UndeliverableHere { get; }
+
+ /// <summary>
+ /// Returns any delivery annotations that were applied to this modified
outcome
+ /// </summary>
+ IReadOnlyDictionary<string, object> MessageAnnotations { get; }
+
+ }
+}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/IRejected.cs
b/src/Proton.Client/Client/IRejected.cs
new file mode 100644
index 0000000..042bddc
--- /dev/null
+++ b/src/Proton.Client/Client/IRejected.cs
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 System.Collections.Generic;
+using Apache.Qpid.Proton.Client.Implementation;
+
+namespace Apache.Qpid.Proton.Client
+{
+ /// <summary>
+ /// Marker interface for Rejected delivery states and outcomes
+ /// </summary>
+ public interface IRejected : IDeliveryState
+ {
+ /// <summary>
+ /// Quick access to any condition value provided in the error condition
sent with the rejected outcome.
+ /// </summary>
+ string Condition { get; }
+
+ /// <summary>
+ /// Quick access to any description value provided in the error
condition sent with the rejected outcome.
+ /// </summary>
+ string Description { get; }
+
+ /// <summary>
+ /// Returns any information entries that were applied to this rejected
outcome in the error condition.
+ /// </summary>
+ IReadOnlyDictionary<string, object> Info { get; }
+
+ }
+}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/IReleased.cs
b/src/Proton.Client/Client/IReleased.cs
new file mode 100644
index 0000000..840ad2c
--- /dev/null
+++ b/src/Proton.Client/Client/IReleased.cs
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 Apache.Qpid.Proton.Client.Implementation;
+
+namespace Apache.Qpid.Proton.Client
+{
+ /// <summary>
+ /// Marker interface for Released delivery states and outcomes
+ /// </summary>
+ public interface IReleased : IDeliveryState
+ {
+ }
+}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/ITransactional.cs
b/src/Proton.Client/Client/ITransactional.cs
new file mode 100644
index 0000000..e782340
--- /dev/null
+++ b/src/Proton.Client/Client/ITransactional.cs
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 Apache.Qpid.Proton.Client.Implementation;
+
+namespace Apache.Qpid.Proton.Client
+{
+ /// <summary>
+ /// Marker interface for Transactional delivery states and outcomes
+ /// </summary>
+ public interface ITransactional : IDeliveryState
+ {
+ /// <summary>
+ /// Returns the outcome that will be applied to the transaction when
discharged.
+ /// </summary>
+ IDeliveryState Outcome { get; }
+ }
+}
\ No newline at end of file
diff --git a/src/Proton.Client/Client/Implementation/ClientDeliveryState.cs
b/src/Proton.Client/Client/Implementation/ClientDeliveryState.cs
index 593a54f..3723f04 100644
--- a/src/Proton.Client/Client/Implementation/ClientDeliveryState.cs
+++ b/src/Proton.Client/Client/Implementation/ClientDeliveryState.cs
@@ -20,6 +20,7 @@ using System.Collections.Generic;
using Apache.Qpid.Proton.Types;
using Apache.Qpid.Proton.Types.Messaging;
using Apache.Qpid.Proton.Types.Transactions;
+using Apache.Qpid.Proton.Types.Transport;
namespace Apache.Qpid.Proton.Client.Implementation
{
@@ -30,6 +31,14 @@ namespace Apache.Qpid.Proton.Client.Implementation
{
public virtual bool IsAccepted => Type == DeliveryStateType.Accepted;
+ public virtual bool IsRejected => Type == DeliveryStateType.Rejected;
+
+ public virtual bool IsReleased => Type == DeliveryStateType.Released;
+
+ public virtual bool IsModified => Type == DeliveryStateType.Modified;
+
+ public virtual bool IsTransactional => Type ==
DeliveryStateType.Transactional;
+
public abstract DeliveryStateType Type { get; }
public abstract Types.Transport.IDeliveryState ProtonDeliveryState {
get; }
@@ -39,12 +48,14 @@ namespace Apache.Qpid.Proton.Client.Implementation
/// <summary>
/// Client version of the proton Accepted delivery state
/// </summary>
- public sealed class ClientAccepted : ClientDeliveryState
+ public sealed class ClientAccepted : ClientDeliveryState, IAccepted
{
public static readonly ClientAccepted Instance = new();
private ClientAccepted() { }
+ public override bool IsAccepted => true;
+
public override DeliveryStateType Type => DeliveryStateType.Accepted;
public override Types.Transport.IDeliveryState ProtonDeliveryState =>
Accepted.Instance;
@@ -54,12 +65,14 @@ namespace Apache.Qpid.Proton.Client.Implementation
/// <summary>
/// Client version of the proton Released delivery state
/// </summary>
- public sealed class ClientReleased : ClientDeliveryState
+ public sealed class ClientReleased : ClientDeliveryState, IReleased
{
public static readonly ClientReleased Instance = new();
private ClientReleased() { }
+ public override bool IsReleased => true;
+
public override DeliveryStateType Type => DeliveryStateType.Released;
public override Types.Transport.IDeliveryState ProtonDeliveryState =>
Released.Instance;
@@ -69,7 +82,7 @@ namespace Apache.Qpid.Proton.Client.Implementation
/// <summary>
/// Client version of the proton Rejected delivery state
/// </summary>
- public sealed class ClientRejected : ClientDeliveryState
+ public sealed class ClientRejected : ClientDeliveryState, IRejected
{
private readonly Rejected rejected = new();
@@ -85,7 +98,7 @@ namespace Apache.Qpid.Proton.Client.Implementation
/// <param name="description">The description value to convey to the
remote</param>
public ClientRejected(string condition, string description)
{
- this.rejected.Error = new Types.Transport.ErrorCondition(condition,
description);
+ rejected.Error = new Types.Transport.ErrorCondition(condition,
description);
}
/// <summary>
@@ -105,14 +118,35 @@ namespace Apache.Qpid.Proton.Client.Implementation
public override DeliveryStateType Type => DeliveryStateType.Rejected;
+ public override bool IsRejected => true;
+
public override Types.Transport.IDeliveryState ProtonDeliveryState =>
rejected;
+ public string Condition => rejected.Error?.Condition?.ToString();
+
+ public string Description => rejected.Error?.Description;
+
+ public IReadOnlyDictionary<string, object> Info =>
ClientConversionSupport.ToStringKeyedMap(rejected.Error?.Info);
+
+ internal static Rejected FromUnknownClientType(IDeliveryState
deliveryState)
+ {
+ if (deliveryState is IRejected rejected)
+ {
+ return new Rejected(new ErrorCondition(rejected.Condition,
+ rejected.Description,
+
ClientConversionSupport.ToSymbolKeyedMap(rejected.Info)));
+ }
+ else
+ {
+ return new Rejected(); // TODO: This loses data from the source
but remains backwards compatible
+ }
+ }
}
/// <summary>
/// Client version of the proton Modified delivery state
/// </summary>
- public sealed class ClientModified : ClientDeliveryState
+ public sealed class ClientModified : ClientDeliveryState, IModified
{
private readonly Modified modified = new();
@@ -120,7 +154,10 @@ namespace Apache.Qpid.Proton.Client.Implementation
{
this.modified.DeliveryFailed = modified.DeliveryFailed;
this.modified.UndeliverableHere = modified.UndeliverableHere;
- this.modified.MessageAnnotations = new Dictionary<Symbol,
object>(modified.MessageAnnotations);
+ if (modified.MessageAnnotations != null)
+ {
+ this.modified.MessageAnnotations = new Dictionary<Symbol,
object>(modified.MessageAnnotations);
+ }
}
/// <summary>
@@ -151,12 +188,31 @@ namespace Apache.Qpid.Proton.Client.Implementation
public override Types.Transport.IDeliveryState ProtonDeliveryState =>
modified;
+ public override bool IsModified => true;
+
+ public bool DeliveryFailed => modified.DeliveryFailed;
+
+ public bool UndeliverableHere => modified.UndeliverableHere;
+
+ public IReadOnlyDictionary<string, object> MessageAnnotations =>
ClientConversionSupport.ToStringKeyedMap(modified.MessageAnnotations);
+
+ internal static Modified FromUnknownClientType(IDeliveryState
deliveryState)
+ {
+ if (deliveryState is IModified modified)
+ {
+ return new Modified(modified.DeliveryFailed,
modified.UndeliverableHere,
ClientConversionSupport.ToSymbolKeyedMap(modified.MessageAnnotations));
+ }
+ else
+ {
+ return new Modified(); // TODO: This loses data from the source
but remains backwards compatible
+ }
+ }
}
/// <summary>
/// Client version of the proton Transactional delivery state
/// </summary>
- public sealed class ClientTransactional : ClientDeliveryState
+ public sealed class ClientTransactional : ClientDeliveryState,
ITransactional
{
private readonly TransactionalState txnState = new();
@@ -168,10 +224,20 @@ namespace Apache.Qpid.Proton.Client.Implementation
public override bool IsAccepted => txnState.Outcome is Accepted;
+ public override bool IsReleased => txnState.Outcome is Released;
+
+ public override bool IsRejected => txnState.Outcome is Rejected;
+
+ public override bool IsModified => txnState.Outcome is Modified;
+
+ public override bool IsTransactional => true;
+
public override DeliveryStateType Type =>
DeliveryStateType.Transactional;
public override Types.Transport.IDeliveryState ProtonDeliveryState =>
txnState;
+ public IDeliveryState Outcome =>
DeliveryStateExtensions.ToClientDeliveryState(txnState.Outcome);
+
}
#region Extension types for Proton and Client delivery state types
@@ -252,8 +318,8 @@ namespace Apache.Qpid.Proton.Client.Implementation
{
DeliveryStateType.Accepted => Accepted.Instance,
DeliveryStateType.Released => Released.Instance,
- DeliveryStateType.Rejected => new Rejected(),// TODO - How do
we aggregate the different values into one DeliveryState Object
- DeliveryStateType.Modified => new Modified(),// TODO - How do
we aggregate the different values into one DeliveryState Object
+ DeliveryStateType.Rejected =>
ClientRejected.FromUnknownClientType(state),
+ DeliveryStateType.Modified =>
ClientModified.FromUnknownClientType(state),
DeliveryStateType.Transactional => throw new
ArgumentException("Cannot manually enlist delivery in AMQP Transactions"),
_ => throw new InvalidOperationException("Client does not
support the given Delivery State type: " + state.Type),
};
diff --git a/src/Proton/Types/Transport/ErrorCondition.cs
b/src/Proton/Types/Transport/ErrorCondition.cs
index d4d6cc1..069898a 100644
--- a/src/Proton/Types/Transport/ErrorCondition.cs
+++ b/src/Proton/Types/Transport/ErrorCondition.cs
@@ -18,12 +18,13 @@
using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
+using Apache.Qpid.Proton.Utilities;
namespace Apache.Qpid.Proton.Types.Transport
{
public sealed class ErrorCondition : ICloneable
{
- public static readonly ulong DescriptorCode = 0x000000000000001dUL;
+ public static readonly ulong DescriptorCode = 0x000000000000001dUL;
public static readonly Symbol DescriptorSymbol =
Symbol.Lookup("amqp:error:list");
public ErrorCondition(string condition) : this(Symbol.Lookup(condition),
null, null) { }
diff --git
a/test/Proton.Client.Tests/Client/Implementation/ClientDeliveryStateTest.cs
b/test/Proton.Client.Tests/Client/Implementation/ClientDeliveryStateTest.cs
new file mode 100644
index 0000000..ffe8b71
--- /dev/null
+++ b/test/Proton.Client.Tests/Client/Implementation/ClientDeliveryStateTest.cs
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 System.Collections.Generic;
+using System.Threading;
+using Apache.Qpid.Proton.Buffer;
+using Apache.Qpid.Proton.Types;
+using Apache.Qpid.Proton.Types.Messaging;
+using Apache.Qpid.Proton.Types.Transactions;
+using Apache.Qpid.Proton.Types.Transport;
+using NUnit.Framework;
+
+namespace Apache.Qpid.Proton.Client.Implementation
+{
+ [TestFixture, Timeout(20000)]
+ public class ClientDeliveryStateTest
+ {
+ [Test]
+ public void TestAcceptedType()
+ {
+ IDeliveryState deliveryState = IDeliveryState.Accepted();
+
+ Assert.NotNull(deliveryState);
+ Assert.IsTrue(deliveryState.IsAccepted);
+ Assert.IsFalse(deliveryState.IsModified);
+ Assert.IsFalse(deliveryState.IsRejected);
+ Assert.IsFalse(deliveryState.IsReleased);
+ Assert.IsFalse(deliveryState.IsTransactional);
+ Assert.AreEqual(DeliveryStateType.Accepted, deliveryState.Type);
+
+ Assert.IsInstanceOf<IAccepted>(deliveryState);
+ }
+
+ [Test]
+ public void TestReleasedType()
+ {
+ IDeliveryState deliveryState = IDeliveryState.Released();
+
+ Assert.NotNull(deliveryState);
+ Assert.IsFalse(deliveryState.IsAccepted);
+ Assert.IsFalse(deliveryState.IsModified);
+ Assert.IsFalse(deliveryState.IsRejected);
+ Assert.IsTrue(deliveryState.IsReleased);
+ Assert.IsFalse(deliveryState.IsTransactional);
+ Assert.AreEqual(DeliveryStateType.Released, deliveryState.Type);
+
+ Assert.IsInstanceOf<IReleased>(deliveryState);
+ }
+
+ [Test]
+ public void TestRejectedType()
+ {
+ IDeliveryState deliveryState = IDeliveryState.Rejected("amqp:error",
"error");
+
+ Assert.NotNull(deliveryState);
+ Assert.IsFalse(deliveryState.IsAccepted);
+ Assert.IsFalse(deliveryState.IsModified);
+ Assert.IsTrue(deliveryState.IsRejected);
+ Assert.IsFalse(deliveryState.IsReleased);
+ Assert.IsFalse(deliveryState.IsTransactional);
+ Assert.AreEqual(DeliveryStateType.Rejected, deliveryState.Type);
+
+ IRejected rejected = deliveryState as IRejected;
+
+ Assert.AreEqual("amqp:error", rejected.Condition);
+ Assert.AreEqual("error", rejected.Description);
+ Assert.IsNull(rejected.Info);
+
+ Assert.IsInstanceOf<IRejected>(deliveryState);
+ }
+
+ [Test]
+ public void TestModifiedType()
+ {
+ IDeliveryState deliveryState = IDeliveryState.Modified(true, true);
+
+ Assert.NotNull(deliveryState);
+ Assert.IsFalse(deliveryState.IsAccepted);
+ Assert.IsTrue(deliveryState.IsModified);
+ Assert.IsFalse(deliveryState.IsRejected);
+ Assert.IsFalse(deliveryState.IsReleased);
+ Assert.IsFalse(deliveryState.IsTransactional);
+ Assert.AreEqual(DeliveryStateType.Modified, deliveryState.Type);
+
+ Assert.IsInstanceOf<IModified>(deliveryState);
+ }
+
+ [Test]
+ public void TestTransactionalWithReleased()
+ {
+ ITransactional state = new
ClientTransactional(CreateTransactional(Released.Instance));
+
+ Assert.IsNotNull(state);
+ Assert.AreEqual(state.Type, DeliveryStateType.Transactional);
+ Assert.IsFalse(state.IsAccepted);
+ Assert.IsFalse(state.IsRejected);
+ Assert.IsTrue(state.IsReleased);
+ Assert.IsFalse(state.IsModified);
+ Assert.IsTrue(state.IsTransactional);
+ Assert.IsTrue(state.Outcome is IReleased);
+ }
+
+ [Test]
+ public void TestTransactionalWithModified()
+ {
+ IDictionary<Symbol, object> symbolicAnnotations = new
Dictionary<Symbol, object>();
+ symbolicAnnotations.Add("test", "value");
+ IDictionary<string, object> annotations = new Dictionary<string,
object>();
+ annotations.Add("test", "value");
+
+ ITransactional state = new
ClientTransactional(CreateTransactional(new Modified(true, true,
symbolicAnnotations)));
+
+ Assert.IsNotNull(state);
+ Assert.AreEqual(state.Type, DeliveryStateType.Transactional);
+ Assert.IsFalse(state.IsAccepted);
+ Assert.IsFalse(state.IsRejected);
+ Assert.IsFalse(state.IsReleased);
+ Assert.IsTrue(state.IsModified);
+ Assert.IsTrue(state.IsTransactional);
+ Assert.IsTrue(state.Outcome is IModified);
+
+ IModified modifiedState = state.Outcome as IModified;
+
+ Assert.IsNotNull(modifiedState);
+ Assert.IsTrue(modifiedState.DeliveryFailed);
+ Assert.IsTrue(modifiedState.UndeliverableHere);
+ Assert.AreEqual(annotations, modifiedState.MessageAnnotations);
+ }
+
+ [Test]
+ public void TestTransactionalWithRejected()
+ {
+ IDictionary<Symbol, object> symbolicInfo = new Dictionary<Symbol,
object>();
+ symbolicInfo.Add("test", "value");
+ IDictionary<string, object> info = new Dictionary<string, object>();
+ info.Add("test", "value");
+
+ ITransactional state = new
ClientTransactional(CreateTransactional(new Rejected(new ErrorCondition("test",
"data", symbolicInfo))));
+
+ Assert.IsNotNull(state);
+ Assert.AreEqual(state.Type, DeliveryStateType.Transactional);
+ Assert.IsFalse(state.IsAccepted);
+ Assert.IsTrue(state.IsRejected);
+ Assert.IsFalse(state.IsReleased);
+ Assert.IsFalse(state.IsModified);
+ Assert.IsTrue(state.IsTransactional);
+ Assert.IsTrue(state.Outcome is IRejected);
+
+ IRejected rejectedState = state.Outcome as IRejected;
+
+ Assert.IsNotNull(rejectedState);
+ Assert.AreEqual("test", rejectedState.Condition);
+ Assert.AreEqual("data", rejectedState.Description);
+ Assert.AreEqual(info, rejectedState.Info);
+ }
+
+ private TransactionalState CreateTransactional(IOutcome outcome)
+ {
+ TransactionalState txnState = new TransactionalState();
+
+ txnState.TxnId = ProtonByteBufferAllocator.Instance.Wrap(new byte[] {
0, 1, 2, 3 });
+ txnState.Outcome = outcome;
+
+ return txnState;
+ }
+ }
+}
\ No newline at end of file
diff --git
a/test/Proton.Client.Tests/Client/Implementation/ClientReceiverTest.cs
b/test/Proton.Client.Tests/Client/Implementation/ClientReceiverTest.cs
index 6059cec..9e578b4 100644
--- a/test/Proton.Client.Tests/Client/Implementation/ClientReceiverTest.cs
+++ b/test/Proton.Client.Tests/Client/Implementation/ClientReceiverTest.cs
@@ -3293,5 +3293,64 @@ namespace Apache.Qpid.Proton.Client.Implementation
peer.WaitForScriptToComplete();
}
}
+
+ [Test]
+ public void TestReceiveMessageAndSendModifiedDispositionToRemote()
+ {
+ byte[] payload = CreateEncodedMessage(new AmqpValue("Hello World"));
+
+ using (ProtonTestServer peer = new ProtonTestServer(loggerFactory))
+ {
+ peer.ExpectSASLAnonymousConnect();
+ peer.ExpectOpen().Respond();
+ peer.ExpectBegin().Respond();
+ peer.ExpectAttach().OfReceiver().Respond();
+ peer.ExpectFlow();
+ peer.RemoteTransfer().WithHandle(0)
+ .WithDeliveryId(0)
+ .WithDeliveryTag(new byte[] { 1 })
+ .WithMore(false)
+ .WithMessageFormat(0)
+ .WithPayload(payload)
+ .Queue();
+ peer.ExpectDisposition().WithFirst(0)
+ .WithSettled(false)
+ .WithState().Modified(true, true);
+ peer.Start();
+
+ string remoteAddress = peer.ServerAddress;
+ int remotePort = peer.ServerPort;
+
+ logger.LogInformation("Test started, peer listening on: {0}:{1}",
remoteAddress, remotePort);
+
+ IClient container = IClient.Create();
+
+ ConnectionOptions options = new ConnectionOptions();
+ ReceiverOptions receiverOptions = new ReceiverOptions()
+ {
+ AutoSettle = false,
+ AutoAccept = false
+ };
+ IConnection connection = container.Connect(remoteAddress,
remotePort, options);
+ IReceiver receiver = connection.OpenReceiver("test-queue",
receiverOptions);
+ IDelivery delivery = receiver.Receive();
+ IMessage<object> message = delivery.Message();
+
+ Assert.IsNotNull(message.Body);
+
+ delivery.Disposition(IDeliveryState.Modified(true, true), false);
+
+ peer.WaitForScriptToComplete();
+ peer.ExpectDetach().Respond();
+ peer.ExpectClose().Respond();
+
+ Assert.IsNotNull(delivery);
+
+ receiver.Close();
+ connection.Close();
+
+ peer.WaitForScriptToComplete();
+ }
+ }
}
}
\ No newline at end of file
diff --git a/test/Proton.Client.Tests/Client/Implementation/ClientSenderTest.cs
b/test/Proton.Client.Tests/Client/Implementation/ClientSenderTest.cs
index b03d4cb..5f6c3fa 100644
--- a/test/Proton.Client.Tests/Client/Implementation/ClientSenderTest.cs
+++ b/test/Proton.Client.Tests/Client/Implementation/ClientSenderTest.cs
@@ -3391,5 +3391,63 @@ namespace Apache.Qpid.Proton.Client.Implementation
peer.WaitForScriptToComplete();
}
}
+
+ [Test]
+ public void TestSendIsRejectedAndTrackerReflectsState()
+ {
+ using (ProtonTestServer peer = new ProtonTestServer(loggerFactory))
+ {
+ peer.ExpectSASLAnonymousConnect();
+ peer.ExpectOpen().Respond();
+ peer.ExpectBegin().Respond();
+ peer.ExpectAttach().OfSender().Respond();
+ peer.RemoteFlow().WithLinkCredit(1).Queue();
+ peer.Start();
+
+ string remoteAddress = peer.ServerAddress;
+ int remotePort = peer.ServerPort;
+
+ logger.LogInformation("Test started, peer listening on: {0}:{1}",
remoteAddress, remotePort);
+
+ IClient container = IClient.Create();
+
+ IConnection connection = container.Connect(remoteAddress,
remotePort).OpenTask.Result;
+ ISession session = connection.OpenSession().OpenTask.Result;
+ SenderOptions options = new SenderOptions()
+ {
+ DeliveryMode = DeliveryMode.AtLeastOnce,
+ AutoSettle = false
+ };
+ ISender sender = session.OpenSender("test-tags",
options).OpenTask.Result;
+
+ peer.WaitForScriptToComplete();
+ peer.ExpectTransfer().WithNonNullPayload()
+ .WithDeliveryTag(new byte[] { 0 })
+ .Respond()
+ .WithSettled(true)
+ .WithState().Rejected("no-space", "disk
full");
+
+ peer.ExpectDetach().Respond();
+ peer.ExpectClose().Respond();
+
+ IMessage<string> message = IMessage<string>.Create("Hello World");
+ ITracker tracker = sender.Send(message);
+
+ Assert.IsNotNull(tracker);
+ Assert.IsNotNull(tracker.SettlementTask.Result);
+ Assert.AreEqual(DeliveryStateType.Rejected,
tracker.RemoteState.Type);
+
+ IRejected rejected = (IRejected)tracker.RemoteState;
+
+ Assert.IsNotNull(rejected);
+ Assert.AreEqual("no-space", rejected.Condition);
+ Assert.AreEqual("disk full", rejected.Description);
+
+ sender.CloseAsync().Wait(TimeSpan.FromSeconds(10));
+ connection.CloseAsync().Wait(TimeSpan.FromSeconds(10));
+
+ peer.WaitForScriptToComplete();
+ }
+ }
}
}
\ No newline at end of file
diff --git
a/test/Proton.Client.Tests/Client/Implementation/ClilentTransactionsTest.cs
b/test/Proton.Client.Tests/Client/Implementation/ClilentTransactionsTest.cs
index 653b631..409b395 100644
--- a/test/Proton.Client.Tests/Client/Implementation/ClilentTransactionsTest.cs
+++ b/test/Proton.Client.Tests/Client/Implementation/ClilentTransactionsTest.cs
@@ -1086,6 +1086,7 @@ namespace Apache.Qpid.Proton.Client.Implementation
Assert.IsNotNull(tracker);
Assert.IsNotNull(tracker.SettlementTask.Result);
+ Assert.IsTrue(tracker.RemoteState.IsTransactional);
Assert.AreEqual(tracker.RemoteState.Type,
DeliveryStateType.Transactional);
Assert.IsNotNull(tracker.State);
Assert.AreEqual(tracker.State.Type,
DeliveryStateType.Transactional,
@@ -1355,6 +1356,7 @@ namespace Apache.Qpid.Proton.Client.Implementation
ITracker tracker =
sender.Send(IMessage<string>.Create("test-message"));
Assert.IsNotNull(tracker.SettlementTask.Result);
+ Assert.IsTrue(tracker.RemoteState.IsTransactional);
Assert.AreEqual(tracker.RemoteState.Type,
DeliveryStateType.Transactional);
try
@@ -1719,6 +1721,12 @@ namespace Apache.Qpid.Proton.Client.Implementation
delivery2.Release();
session.CommitTransaction();
+
+ Assert.IsFalse(delivery1.State.IsReleased);
+ Assert.IsTrue(delivery1.State.IsAccepted);
+ Assert.IsTrue(delivery2.State.IsReleased);
+ Assert.IsFalse(delivery2.State.IsAccepted);
+
receiver.Close();
connection.Close();
@@ -1769,6 +1777,7 @@ namespace Apache.Qpid.Proton.Client.Implementation
Assert.IsNotNull(tracker);
Assert.IsNotNull(tracker.AwaitAccepted());
Assert.IsTrue(tracker.RemoteState.IsAccepted);
+ Assert.IsTrue(tracker.RemoteState.IsTransactional);
Assert.AreEqual(tracker.RemoteState.Type,
DeliveryStateType.Transactional,
"Delivery inside transaction should have
Transactional state");
Assert.AreEqual(tracker.State.Type,
DeliveryStateType.Transactional,
@@ -1783,5 +1792,78 @@ namespace Apache.Qpid.Proton.Client.Implementation
peer.WaitForScriptToComplete();
}
}
+
+ [Test]
+ public void TestModifiedDispositionInTransaction()
+ {
+ byte[] txnId = new byte[] { 0, 1, 2, 3 };
+ byte[] payload = CreateEncodedMessage(new AmqpValue("Hello World"));
+
+ using (ProtonTestServer peer = new ProtonTestServer(loggerFactory))
+ {
+ peer.ExpectSASLAnonymousConnect();
+ peer.ExpectOpen().Respond();
+ peer.ExpectBegin().Respond();
+ peer.ExpectAttach().OfReceiver().Respond();
+ peer.ExpectFlow();
+ peer.Start();
+
+ string remoteAddress = peer.ServerAddress;
+ int remotePort = peer.ServerPort;
+
+ logger.LogInformation("Test started, peer listening on: {0}:{1}",
remoteAddress, remotePort);
+
+ IClient container = IClient.Create();
+ IConnection connection = container.Connect(remoteAddress,
remotePort);
+ ISession session = connection.OpenSession();
+ ReceiverOptions options = new ReceiverOptions()
+ {
+ AutoAccept = false,
+ AutoSettle = false
+ };
+ IReceiver receiver = session.OpenReceiver("test-queue",
options).OpenTask.Result;
+
+ peer.ExpectCoordinatorAttach().Respond();
+ peer.RemoteFlow().WithLinkCredit(2).Queue();
+ peer.ExpectDeclare().Accept(txnId);
+ peer.RemoteTransfer().WithHandle(0)
+ .WithDeliveryId(0)
+ .WithDeliveryTag(new byte[] { 1 })
+ .WithMore(false)
+ .WithMessageFormat(0)
+ .WithPayload(payload).Queue();
+ peer.ExpectDisposition().WithSettled(true)
+ .WithState()
+ .Transactional()
+ .WithTxnId(txnId)
+ .WithModified(true, true);
+
+ peer.ExpectDischarge().WithFail(false).WithTxnId(txnId).Accept();
+ peer.ExpectDetach().Respond();
+ peer.ExpectClose().Respond();
+
+ session.BeginTransaction();
+
+ IDelivery delivery = receiver.Receive(TimeSpan.FromSeconds(1));
+
+ Assert.IsNotNull(delivery);
+ Assert.IsFalse(delivery.Settled);
+ Assert.IsNull(delivery.State);
+
+ delivery.Modified(true, true);
+
+ session.CommitTransaction();
+ receiver.CloseAsync();
+ connection.CloseAsync();
+
+ Assert.IsTrue(delivery.State is ITransactional);
+
+ ITransactional txn = (ITransactional)delivery.State;
+
+ Assert.IsTrue(txn.Outcome is IModified);
+
+ peer.WaitForScriptToComplete();
+ }
+ }
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
