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.
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.