Skip to content

Commit 6b42230

Browse files
zhenlineozhenlineo
authored
Merge pull request #218 from ali-ince/1.5-async-perf-improvement
1.5 async performance improvement
2 parents 106cd98 + 4b80d54 commit 6b42230

File tree

5 files changed

+111
-231
lines changed

5 files changed

+111
-231
lines changed

‎Neo4j.Driver/Neo4j.Driver.Tests/IO/ChunkReaderTests.cs‎

Lines changed: 5 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,8 @@ public void ShouldCleanupChunkBufferIfItExceedsMaxChunkBufferSize()
167167
real.Should().HaveCount(ushort.MaxValue);
168168
real.Should().Contain(0);
169169

170-
chunkBuffer.Length.Should().Be(2);
170+
chunkBuffer.Length.Should().Be(0);
171+
chunkBuffer.Position.Should().Be(0);
171172
}
172173

173174
[Fact]
@@ -188,109 +189,11 @@ public async void ShouldCleanupChunkBufferIfItExceedsMaxChunkBufferSizeAsync()
188189
real.Should().HaveCount(ushort.MaxValue);
189190
real.Should().Contain(0);
190191

191-
chunkBuffer.Length.Should().Be(2);
192+
chunkBuffer.Length.Should().Be(0);
193+
chunkBuffer.Position.Should().Be(0);
192194
}
193195

194196
}
195-
196-
//public class ReadBSyteMethod
197-
//{
198-
// [Theory]
199-
// [InlineData(new byte[] {0x00, 0x01, 0x80, 0x00, 0x00}, sbyte.MinValue)]
200-
// [InlineData(new byte[] {0x00, 0x01, 0x7F, 0x00, 0x00}, sbyte.MaxValue)]
201-
// [InlineData(new byte[] {0x00, 0x01, 0x00, 0x00, 0x00}, 0)]
202-
// [InlineData(new byte[] {0x00, 0x01, 0xFF, 0x00, 0x00}, -1)]
203-
// public void ShouldReturnTheCorrectValue(byte[] response, sbyte correctValue)
204-
// {
205-
// var chunkedInput = IOExtensions.CreateChunkedPackStreamReaderFromBytes(response);
206-
// var actual = chunkedInput.NextSByte();
207-
// actual.Should().Be(correctValue); //, $"Got: {actual}, expected: {correctValue}");
208-
// }
209-
//}
210-
211-
//public class MultipleChunksTests
212-
//{
213-
// private readonly ITestOutputHelper _output;
214-
215-
// public MultipleChunksTests(ITestOutputHelper output)
216-
// {
217-
// _output = output;
218-
// }
219-
220-
// [Theory]
221-
// //-----------------------|---head1--|----|---head2---|-----------|--msg end--|
222-
// [InlineData(new byte[] { 0x00, 0x01, 0x00, 0x00, 0x02, 0x01, 0x02, 0x00, 0x00 }, new byte[] { 0x00, 0x01, 0x02})]
223-
// public void ShouldReadMessageAcrossChunks(byte[] input, byte[] correctValue)
224-
// {
225-
// var chunkedInput = IOExtensions.CreateChunkedPackStreamReaderFromBytes(input);
226-
227-
// byte[] actual = chunkedInput.ReadBytes(3);
228-
229-
// actual.Should().Equal(correctValue);
230-
// }
231-
232-
// [Theory]
233-
// //-----------------------|---head1--|----|---head2---|-----------|--msg end--|
234-
// [InlineData(new byte[] { 0x00, 0x01, 0x00, 0x00, 0x02, 0x01, 0x02, 0x00, 0x00 }, new byte[] { 0x00, 0x01, 0x02 })]
235-
// public void ShouldLogBytes(byte[] input, byte[] correctValue)
236-
// {
237-
// var loggerMock = new Mock<ILogger>();
238-
// loggerMock.Setup(x => x.Trace(It.IsAny<string>(), It.IsAny<object[]>(), It.IsAny<int>(), It.IsAny<int>()))
239-
// .Callback<string, object[]>((s, o) => _output.WriteLine(s + ((byte[])o[0]).ToHexString(showX: true)));
240-
241-
// var chunkedInput = IOExtensions.CreateChunkedPackStreamReaderFromBytes(input, loggerMock.Object);
242-
243-
// byte[] actual = chunkedInput.ReadBytes(3);
244-
// actual.Should().Equal(correctValue);
245-
// loggerMock.Verify(x => x.Trace("S: ", It.IsAny<byte[]>(), It.IsAny<int>(), It.IsAny<int>()), Times.AtLeastOnce);
246-
// }
247-
248-
// [Theory]
249-
// //-----------------------|---head1--|----|---head2---|-----------|--msg end--|
250-
// [InlineData(new byte[] { 0x00, 0x01, 0x00, 0x00, 0x02, 0x01, 0x02, 0x00, 0x00 }, new byte[] { 0x00, 0x01, 0x02 })]
251-
// public void ShouldReadMessageBiggerThanChunkSize(byte[] input, byte[] correctValue)
252-
// {
253-
// var chunkedInput = IOExtensions.CreateChunkedPackStreamReaderFromBytes(input);
254-
255-
// byte[] actual = chunkedInput.ReadBytes(3);
256-
// actual.Should().Equal(correctValue);
257-
// }
258-
//}
259-
260-
//public class ChunkHeaderTests
261-
//{
262-
// private readonly Random _random = new Random();
263-
// public byte Getbyte()
264-
// {
265-
// var num = _random.Next(0, 26); // 0 to 25
266-
// byte letter = (byte)('a' + num);
267-
// return letter;
268-
// }
269-
270-
// [Fact]
271-
// public void ShouldReadHeaderWithinUnsignedShortRange()
272-
// {
273-
// for (var i = 1; i <= UInt16.MaxValue; i = (i << 1) + 1) // i: [0x1, 0xFFFF]
274-
// {
275-
// ushort chunkHeaderSize = (ushort)(i & 0xFFFF);
276-
277-
// var input = new byte[chunkHeaderSize + 2 + 2]; // 0xXX, 0xXX, ..., 0x00, 0x00
278-
// input[0] = (byte)((chunkHeaderSize & 0xFF00) >> 8);
279-
// input[1] = (byte)(chunkHeaderSize & 0xFF);
280-
// for (int j = 2; j < chunkHeaderSize + 2; j++)
281-
// {
282-
// input[j] = Getbyte();
283-
// }
284-
285-
// var chunkedInput = IOExtensions.CreateChunkedPackStreamReaderFromBytes(input);
286-
// byte[] actual = chunkedInput.ReadBytes(chunkHeaderSize);
287-
// for (int j = 0; j < actual.Length; j++)
288-
// {
289-
// actual[j].Should().Be(input[2 + j]);
290-
// }
291-
// }
292-
// }
293-
//}
294-
197+
295198
}
296199
}

‎Neo4j.Driver/Neo4j.Driver/Internal/Connector/IInputStream.cs‎

Lines changed: 0 additions & 41 deletions
This file was deleted.

‎Neo4j.Driver/Neo4j.Driver/Internal/Connector/IOutputStream.cs‎

Lines changed: 0 additions & 31 deletions
This file was deleted.

‎Neo4j.Driver/Neo4j.Driver/Internal/IO/BoltReader.cs‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ public void Read(IMessageResponseHandler responseHandler)
7070

7171
_chunkReader.ReadNextMessage(_bufferStream);
7272

73-
ProcessMessage(responseHandler);
73+
ConsumeMessages(responseHandler);
7474
}
7575

7676
public Task ReadAsync(IMessageResponseHandler responseHandler)
@@ -81,14 +81,22 @@ public Task ReadAsync(IMessageResponseHandler responseHandler)
8181
_chunkReader.ReadNextMessageAsync(_bufferStream)
8282
.ContinueWith(t =>
8383
{
84-
ProcessMessage(responseHandler);
84+
ConsumeMessages(responseHandler);
8585
}, TaskContinuationOptions.ExecuteSynchronously);
8686
}
8787

88-
private void ProcessMessage(IMessageResponseHandler responseHandler)
88+
private void ConsumeMessages(IMessageResponseHandler responseHandler)
8989
{
9090
_bufferStream.Position = 0;
9191

92+
while (_bufferStream.Length > _bufferStream.Position)
93+
{
94+
ProcessMessage(responseHandler);
95+
}
96+
}
97+
98+
private void ProcessMessage(IMessageResponseHandler responseHandler)
99+
{
92100
var structure = (PackStreamStruct)_packStreamReader.Read();
93101

94102
switch (structure.Signature)

0 commit comments

Comments
 (0)