[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558418524



##
File path: src/DotPulsar/Internal/MessageQueue.cs
##
@@ -0,0 +1,53 @@
+/*
+ * 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 Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+public sealed class MessageQueue : IMessageQueue, 
IDequeue, IDisposable

Review comment:
   Done: ddfd374

##
File path: src/DotPulsar/Internal/Abstractions/IUnackedMessageTracker.cs
##
@@ -0,0 +1,30 @@
+/*
+ * 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.Abstractions
+{
+using DotPulsar.Abstractions;
+using System.Threading.Tasks;
+using System.Threading;
+using System;
+
+public interface IUnackedMessageTracker : IDisposable
+{
+void Add(MessageId messageId);
+
+void Ack(MessageId messageId);

Review comment:
   Done: ddfd374





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558418667



##
File path: src/DotPulsar/Internal/UnackedMessageTracker.cs
##
@@ -0,0 +1,115 @@
+/*
+ * 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 Abstractions;
+using DotPulsar.Abstractions;
+using System;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Linq;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+public readonly struct AwaitingAck
+{
+public MessageId MessageId { get; }
+public long Timestamp { get; }
+
+public AwaitingAck(MessageId messageId)
+{
+MessageId = messageId;
+Timestamp = Stopwatch.GetTimestamp();
+}
+
+public TimeSpan Elapsed => TimeSpan.FromTicks(
+(long) ((Stopwatch.GetTimestamp() - Timestamp) / (double) 
Stopwatch.Frequency * TimeSpan.TicksPerSecond));
+}
+
+public sealed class UnackedMessageTracker : IUnackedMessageTracker
+{
+private readonly TimeSpan _ackTimeout;
+private readonly TimeSpan _pollingTimeout;
+private readonly ConcurrentQueue _awaitingAcks;
+private readonly List _acked;
+private readonly CancellationTokenSource _cancellationTokenSource;
+
+public UnackedMessageTracker(TimeSpan ackTimeout, TimeSpan 
pollingTimeout)
+{
+_ackTimeout = ackTimeout;
+_pollingTimeout = pollingTimeout;
+_awaitingAcks = new ConcurrentQueue();
+_acked = new List();
+_cancellationTokenSource = new CancellationTokenSource();
+}
+
+public void Add(MessageId messageId)
+{
+_awaitingAcks.Enqueue(new AwaitingAck(messageId));
+}
+
+public void Ack(MessageId messageId)
+{
+// We only need to store the highest cumulative ack we see (if 
there is one)
+// and the MessageIds not included by that cumulative ack.
+_acked.Add(messageId);
+}
+
+public Task Start(IConsumer consumer, CancellationToken 
cancellationToken = default)
+{
+CancellationToken token =
+  CancellationTokenSource.CreateLinkedTokenSource(
+  _cancellationTokenSource.Token, cancellationToken).Token;
+
+return Task.Run(async () =>
+{
+while (!token.IsCancellationRequested)
+{
+var messages = CheckUnackedMessages();
+
+if (messages.Count() > 0)
+await 
consumer.RedeliverUnacknowledgedMessages(messages, token);
+
+await Task.Delay(_pollingTimeout, token);
+}
+}, token);
+}
+
+private IEnumerable CheckUnackedMessages()
+{
+var result = new List();
+
+while (_awaitingAcks.TryPeek(out AwaitingAck awaiting)
+&& awaiting.Elapsed > _ackTimeout)
+{
+if (_awaitingAcks.TryDequeue(out awaiting))
+{
+if (!_acked.Contains(awaiting.MessageId))
+result.Add(awaiting.MessageId);
+else
+_acked.Remove(awaiting.MessageId);
+}
+}
+
+return result;
+}
+
+public void Dispose()
+{
+this._cancellationTokenSource.Cancel();

Review comment:
   Done: ddfd374

##
File path: src/DotPulsar/Internal/UnackedMessageTracker.cs
##
@@ -0,0 +1,115 @@
+/*
+ * 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 Abstractions;
+using DotPulsar.Abstractions;
+using System;
+using System.Colle

[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558418794



##
File path: tests/DotPulsar.Tests/Internal/MessageAcksTrackerTests.cs
##
@@ -0,0 +1,158 @@
+/*
+ * 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.Tests.Internal
+{
+using AutoFixture;
+using AutoFixture.AutoNSubstitute;
+using DotPulsar.Abstractions;
+using DotPulsar.Internal;
+using NSubstitute;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Linq.Expressions;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+public class UnackedMessageTrackerTests
+{
+
+[Fact]
+public async void Start_GivenAMessageIdIsNotAcked_ShouldRedeliver()
+{
+//Arrange
+var fixture = new Fixture();
+fixture.Customize(new AutoNSubstituteCustomization());
+var consumer = Substitute.For();
+var messageId = MessageId.Latest;
+var cts = new CancellationTokenSource();
+
+
+var tracker = new UnackedMessageTracker(
+TimeSpan.FromMilliseconds(10),
+TimeSpan.FromMilliseconds(1));
+
+//Act
+tracker.Add(messageId);
+cts.CancelAfter(20);
+try { await tracker.Start(consumer, cts.Token); }
+catch (TaskCanceledException) { }
+
+//Assert
+await consumer
+.Received(1)
+.RedeliverUnacknowledgedMessages(
+Arg.Is(EquivalentTo(new List() { messageId })),
+Arg.Any());
+}
+
+[Fact]
+public async void 
Start_GivenAMessageIdIsAckedWithinTimeout_ShouldNotRedeliver()
+{
+//Arrange
+var fixture = new Fixture();
+fixture.Customize(new AutoNSubstituteCustomization());
+var consumer = Substitute.For();
+var messageId = MessageId.Latest;
+var cts = new CancellationTokenSource();
+
+
+var tracker = new UnackedMessageTracker(
+TimeSpan.FromMilliseconds(10),
+TimeSpan.FromMilliseconds(1));
+
+//Act
+tracker.Add(messageId);
+cts.CancelAfter(20);
+var _ = Task.Delay(5).ContinueWith(_ => tracker.Ack(messageId));
+try { await tracker.Start(consumer, cts.Token); }
+catch (TaskCanceledException) { }
+
+//Assert
+await consumer
+.DidNotReceive()
+.RedeliverUnacknowledgedMessages(
+Arg.Any>(),
+Arg.Any());
+}
+
+[Fact]
+public async void 
Start_GivenAMessageIdIsNotAckedWithinTimeout_ShouldRedeliver()
+{
+//Arrange
+var fixture = new Fixture();
+fixture.Customize(new AutoNSubstituteCustomization());
+var consumer = Substitute.For();
+var messageId = MessageId.Latest;
+var cts = new CancellationTokenSource();
+
+
+var tracker = new UnackedMessageTracker(
+TimeSpan.FromMilliseconds(10),
+TimeSpan.FromMilliseconds(1));
+
+//Act
+tracker.Add(messageId);
+cts.CancelAfter(20);
+
+var _ = Task.Delay(15).ContinueWith(_ => tracker.Ack(messageId));
+try { await tracker.Start(consumer, cts.Token); }
+catch (TaskCanceledException) { }
+
+//Assert
+await consumer
+.Received(1)
+.RedeliverUnacknowledgedMessages(
+Arg.Any>(),
+Arg.Any());
+}
+
+[Fact]
+public async void 
Start_GivenAMessageIdIsNotAckedWithinTimeout_ShouldRedeliverOnlyOnce()
+{
+//Arrange
+var fixture = new Fixture();
+fixture.Customize(new AutoNSubstituteCustomization());
+var consumer = Substitute.For();
+var messageId = MessageId.Latest;
+var cts = new CancellationTokenSource();
+

Review comment:
   Done: ddfd374

##
File path: tests/DotPulsar.Tests/Internal/MessageAcksTrackerTests.cs
##
@@ -0,0 +1,158 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not us

[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558433871



##
File path: src/DotPulsar/Internal/InactiveUnackedMessageTracker.cs
##
@@ -0,0 +1,31 @@
+namespace DotPulsar.Internal
+{
+using System.Threading;

Review comment:
   Ok, I clearly don't have that in my version, strange I'll have a look to 
see how to configure my editor to respect .editconfig





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558434707



##
File path: src/DotPulsar/Internal/ConsumerChannel.cs
##
@@ -108,6 +108,9 @@ public async Task Send(CommandAck command, 
CancellationToken cancellationToken)
 }
 
 command.ConsumerId = _id;
+
+_queue.Acknowledge(new MessageId(messageId));

Review comment:
   Ok I addressed the MessageId issue by implementing the right interfaces 
for comparison here 8841930, I'm not sure about the extension pattern for 
MessageIdData though (using partials). I refactored the use of MessageIdData in 
ad5827114c8d019aac86d300818b718ae8ab9d19 all tests pass.  





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [pulsar-dotpulsar] dionjansen commented on a change in pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on a change in pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#discussion_r558440605



##
File path: src/DotPulsar/Internal/UnackedMessageTracker.cs
##
@@ -0,0 +1,101 @@
+namespace DotPulsar.Internal
+{
+using Abstractions;
+using DotPulsar.Abstractions;
+using System;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Linq;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+public readonly struct AwaitingAck
+{
+public MessageId MessageId { get; }
+public long Timestamp { get; }
+
+public AwaitingAck(MessageId messageId)
+{
+MessageId = messageId;
+Timestamp = Stopwatch.GetTimestamp();
+}
+
+public TimeSpan Elapsed => TimeSpan.FromTicks(
+(long) ((Stopwatch.GetTimestamp() - Timestamp) / 
(double)Stopwatch.Frequency * TimeSpan.TicksPerSecond));
+}
+
+public sealed class UnackedMessageTracker : IUnackedMessageTracker
+{
+private readonly TimeSpan _ackTimeout;
+private readonly TimeSpan _pollingTimeout;
+private readonly ConcurrentQueue _awaitingAcks;
+private readonly List _acked;
+private readonly CancellationTokenSource _cancellationTokenSource;
+
+
+public UnackedMessageTracker(TimeSpan ackTimeout, TimeSpan 
pollingTimeout)
+{
+_ackTimeout = ackTimeout;
+_pollingTimeout = pollingTimeout;
+_awaitingAcks = new ConcurrentQueue();
+_acked = new List();
+_cancellationTokenSource = new CancellationTokenSource();
+}
+
+public void Add(MessageId messageId)
+{
+_awaitingAcks.Enqueue(new AwaitingAck(messageId));
+}
+
+public void Ack(MessageId messageId)
+{
+// We only need to store the highest cumulative ack we see (if 
there is one)
+// and the MessageIds not included by that cumulative ack.
+_acked.Add(messageId);
+}
+
+public Task Start(IConsumer consumer, CancellationToken 
cancellationToken = default)
+{
+CancellationToken token =
+  CancellationTokenSource.CreateLinkedTokenSource(
+  _cancellationTokenSource.Token, cancellationToken).Token;
+
+return Task.Run(async () => {

Review comment:
   Ok fixed in f02166b I followed the same pattern as in StateMonitor.





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [pulsar-dotpulsar] dionjansen commented on pull request #67: Support for (optional) ack timeout and nack delay for consumers

[GitBox]

dionjansen commented on pull request #67:
URL: https://github.com/apache/pulsar-dotpulsar/pull/67#issuecomment-761063358


   @blankensteiner sorry for the delay, I addressed all your remarks, let me 
know what you think!



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[Success Story] How Pulsar Helps Iterable Scale its Customer Engagement Platform

[anonymitaet _]

Hi Pulsar enthusiasts,

Hope you are doing well.

Recently, we have a fresh Pulsar use case:

Iterable had been using RabbitMQ heavily and relied on its features to handle 
internal messaging. The transition to Pulsar has been interesting and sometimes 
challenging, but quite successful so far.

Curious about how Pulsar supports a wide feature set that makes it a viable 
alternative to many other distributed messaging technologies currently being 
used in Iterable’s architecture?

Check it 
out!
[cid:image001.jpg@01D6EB1F.E0A76A70]
Sincerely,
Anonymitaet