Reputation: 1367
I have following code:
public class Batcher<TPayload> : IBatcher<TPayload>
{
private static readonly BufferBlock<BatchElement<TPayload>> BufferBlock = new BufferBlock<BatchElement<TPayload>>(new DataflowBlockOptions
{
EnsureOrdered = true
});
private readonly TransformBlock<BatchElement<TPayload>, BatchElement<TPayload>> BufferInterceptor;
private readonly TransformBlock<BatchElement<TPayload>, BatchElement<TPayload>> TimeoutInterceptor;
public EventsBatcher(int size, int interval, IMagicService magicService, ILogger<Batcher<TPayload, TStrategy>> logger)
{
BufferInterceptor =
new TransformBlock<BatchElement<TPayload>, BatchElement<TPayload>>(x =>
{
logger.LogInformation($"Get a message with value: {x}");
return x;
});
TimeoutInterceptor =
new TransformBlock<BatchElement<TPayload>, BatchElement<TPayload>>(x =>
{
logger.LogInformation($"Move out from transformation block with a value: {x}");
return x;
});
var batchBlock = new BatchBlock<BatchElement<TPayload>>(size, new GroupingDataflowBlockOptions()
{
EnsureOrdered = true
});
var timer = new Timer(async _ =>
{
try
{
batchBlock.TriggerBatch();
var data = await batchBlock.ReceiveAsync();
if (!data.Any() && data.SomeLogic())
return;
await magicService.PushMessageAsync(batchElement.Payload);
}
catch (Exception e)
{
logger.LogError($"Error occurs while trying to invoke action on batch", e);
}
}, null, 0, 500);
var timeoutBlock = new TransformBlock<BatchElement<TPayload>, BatchElement<TPayload>>(v =>
{
timer.Change(interval, Timeout.Infinite);
return v;
});
TimeoutInterceptor.LinkTo(batchBlock);
timeoutBlock.LinkTo(TimeoutInterceptor);
BufferInterceptor.LinkTo(timeoutBlock);
BufferBlock.LinkTo(BufferInterceptor);
}
public async Task<Result<Unit>> SendAsync(BatchElement<TPayload> msg, CancellationToken token = new CancellationToken())
{
try
{
var result = await BufferBlock.SendAsync(msg, token);
return result
? ResultFactory.CreateSuccess()
: ResultFactory.CreateFailure<Unit>("Message was refused by queue");
}
catch (Exception e)
{
return ResultFactory.CreateFailure<Unit>(e.Message);
}
}
}
Which responsibility is to evaluate somehow data every x milliseconds. I try to write unit tests to that to be sure that everything works fine. Those tests are here:
public class BatcherTests
{
public EventsBatcher<int> Initialize(Dictionary<DateTime, int> output)
{
var busMock = new Mock<IMagicService>();
busMock.Setup(x => x.PushMessageAsync(It.IsAny<int>()))
.Callback<Data>((data) =>
{
output.Add(DateTime.Now, data);
}).Returns(Task.CompletedTask);
var loggerMock = new Mock<ILogger<Batcher<int>>>();
return new Batcher<int>(
2,
5000,
busMock.Object,
loggerMock.Object
);
}
[Fact]
public async Task Batcher_ShouldRemoveDuplicatedMessages()
{
var output = new Dictionary<DateTime, int>();
var batcher = Initialize(output);
var first = await batcher.SendAsync(new MockEvent { Payload = 1 });
var second = await batcher.SendAsync(new MockEvent { Payload = 1 });
(first.IsSuccess && second.IsSuccess).ShouldBeTrue();
while (output.Count != 2)
{
}
output.Count.ShouldBe(2);
output.First().Value.ShouldBe(1);
output.Last().Value.ShouldBe(1);
output.Clear();
}
[Fact]
public async Task Batcher_WhenSizeIsSetTo2AndWeSend3Items_ReturnTwoBatchedItemsWithDateIntervalPlusMinus5000msAndAllSendRequestsEndsWithSuccess()
{
var output = new Dictionary<DateTime, int>();
var batcher = Initialize(output);
var first = await batcher.SendAsync(new MockEvent { Payload = 1 });
var second = await batcher.SendAsync(new MockEvent { Payload = 1 });
var third = await batcher.SendAsync(new MockEvent { Payload = 1 });
(first.IsSuccess && second.IsSuccess && third.IsSuccess).ShouldBeTrue();
while (output.Count != 2) //never ends because there are already two elements in output dictionary
{
}
output.Count.ShouldBe(2);
output.First().Value.ShouldBe(2);
output.Last().Value.ShouldBe(1);
var interval = (output.Last().Key - output.First().Key).TotalSeconds;
(interval >= 4.5d && interval <= 5.5d).ShouldBeTrue();
output.Clear();
}
}
But the strange thing is that when I run them separately they end up with a success status. But when I run them all together one of them seems to stuck. This is because a dictionary which is passed to a logic method has 2 elements inside while starting a test. I don't see here a possibility of shared context since stub
class is created at the beginning of test cases, the same with a dictionary. Is there something that I missing? I also try to split those test cases to separe classes but the same behavior occurs.
Upvotes: 0
Views: 119
Reputation: 29840
There is shared stated, but it is not in the test (directly).
Your BufferBlock
is declared as static
in the class Batcher<TPayload>
. There is your shared state.
private static readonly BufferBlock<BatchElement<TPayload>> BufferBlock = new BufferBlock<BatchElement<TPayload>>(new DataflowBlockOptions
{
EnsureOrdered = true
});
When multiple tests are executed that shared block is linked to the other blocks multiple times.
Upvotes: 1