-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #20 from eventuate-tram/Improvements
Get public repo up-to-date with latest improvements
- Loading branch information
Showing
69 changed files
with
1,556 additions
and
374 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,3 @@ | ||
| # Auto detect text files and perform LF normalization | ||
| * text=auto | ||
| * text=auto | ||
| *.sh text eol=lf |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| # Changelog | ||
|
|
||
| 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 | ||
| - Initial port of Eventuate Tram in .NET supporting the following functionality: | ||
| - messaging - send and receive messages over named channels | ||
| - events - publish domain events and subscribe to domain events | ||
|
|
||
| [Unreleased]: https://github.com/eventuate-tram/eventuate-tram-core-dotnet/commits/HEAD |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| Set-Item -Path Env:CDC_SERVICE_DOCKER_VERSION -Value ("0.6.0.RELEASE") | ||
| Push-Location -Path IO.Eventuate.Tram.IntegrationTests | ||
|
|
||
| docker compose down --remove-orphans | ||
| docker compose up -d mssql | ||
| docker compose up -d zookeeper | ||
| docker compose up -d kafka | ||
|
|
||
| Start-Sleep -Seconds 40 | ||
| docker stats --no-stream | ||
|
|
||
| docker-compose up --exit-code-from kafka-setup kafka-setup | ||
| docker compose up --exit-code-from dbsetup dbsetup | ||
| docker compose up -d cdcservice | ||
|
|
||
| docker compose up -d kafka-ui | ||
|
|
||
| Pop-Location |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
91 changes: 91 additions & 0 deletions
91
IO.Eventuate.Tram.IntegrationTests/TestFixtures/BackpressureTests.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using IO.Eventuate.Tram.Events.Common; | ||
| using IO.Eventuate.Tram.IntegrationTests.TestHelpers; | ||
| using IO.Eventuate.Tram.Local.Kafka.Consumer; | ||
| using NUnit.Framework; | ||
|
|
||
| namespace IO.Eventuate.Tram.IntegrationTests.TestFixtures; | ||
|
|
||
| public class BackpressureTests : IntegrationTestsBase | ||
| { | ||
| private const uint PauseThreshold = 20; | ||
| private const uint ResumeThreshold = 5; | ||
|
|
||
| [SetUp] | ||
| public async Task Setup() | ||
| { | ||
| await CleanupKafkaTopicsAsync(); | ||
| // Initialize the backpressure properties | ||
| var properties = new EventuateKafkaConsumerConfigurationProperties | ||
| { | ||
| BackPressure = | ||
| { | ||
| PauseThreshold = PauseThreshold, | ||
| ResumeThreshold = ResumeThreshold | ||
| } | ||
| }; | ||
| TestSetup("eventuate", false, properties); | ||
| await CleanupTestAsync(); | ||
| } | ||
|
|
||
| [TearDown] | ||
| public void TearDown() | ||
| { | ||
| DisposeTestHost(); | ||
| } | ||
|
|
||
| [Test] | ||
| public void PublishWithBackpressure_Send1000Messages_AllMessagesEventuallyProcessed() | ||
| { | ||
| // Arrange | ||
| uint messagesPerType = 250; | ||
| uint totalMessages = messagesPerType * 4; | ||
| TestMessageType1 msg1 = new TestMessageType1("Msg1", 1, 1.2); | ||
| TestMessageType2 msg2 = new TestMessageType2("Msg2", 2); | ||
| TestMessageType3 msg3 = new TestMessageType3("Msg3", 3); | ||
| TestMessageType4 msg4 = new TestMessageType4("Msg4", 4); | ||
|
|
||
| for (int x = 0; x < messagesPerType; x++) | ||
| { | ||
| GetTestPublisher().Publish(AggregateType12, AggregateType12, new List<IDomainEvent> { msg1 }); | ||
| GetTestPublisher().Publish(AggregateType12, AggregateType12, new List<IDomainEvent> { msg2 }); | ||
| GetTestPublisher().Publish(AggregateType34, AggregateType34, new List<IDomainEvent> { msg3 }); | ||
| GetTestPublisher().Publish(AggregateType34, AggregateType34, new List<IDomainEvent> { msg4 }); | ||
| } | ||
|
|
||
| // Act | ||
| TestEventConsumer consumer = GetTestConsumer(); | ||
|
|
||
| // Allow time for messages to process | ||
| int count = 300; | ||
| while (consumer.TotalMessageCount() < totalMessages && count > 0) | ||
| { | ||
| TestContext.WriteLine($"TotalMessageCount: {consumer.TotalMessageCount()} ({count})"); | ||
| Thread.Sleep(1000); | ||
| count--; | ||
| } | ||
| ShowTestResults(); | ||
|
|
||
| // Assert | ||
| Assert.That(GetDbContext().Messages.Count(), | ||
| Is.EqualTo(totalMessages), "Number of messages produced"); | ||
| Assert.That(GetDbContext().Messages.Count(msg => msg.Published == 0), | ||
| Is.EqualTo(0), "Number of unpublished messages"); | ||
| foreach (var eventType in new[] { typeof(TestMessageType1), typeof(TestMessageType2), typeof(TestMessageType3), typeof(TestMessageType4) }) | ||
| { | ||
| TestEventConsumer.EventStatistics eventStatistics = consumer.GetEventStatistics(eventType); | ||
| Assert.That(eventStatistics.MessageCount, | ||
| Is.EqualTo(messagesPerType), $"Number of {eventType.Name} messages received by consumer"); | ||
| Assert.That(eventStatistics.ReceivedMessages.Count, | ||
| Is.EqualTo(messagesPerType), $"Number of received {eventType.Name} messages"); | ||
| } | ||
|
|
||
| Assert.That(consumer.TotalMessageCount(), | ||
| Is.EqualTo(totalMessages), "Total number of messages received by consumer"); | ||
| Assert.That(GetDbContext().ReceivedMessages.Count(msg => msg.MessageId != null), | ||
| Is.EqualTo(totalMessages), "Number of received messages"); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.