Skip to content

[WIP] Akka.IO / Akka.Streams: ByteString rewrite #7136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Aaronontheweb wants to merge 20 commits into
base: dev
Choose a base branch
from
Draft

[WIP] Akka.IO / Akka.Streams: ByteString rewrite #7136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
96a6d11
replace `ByteString` internals with `ReadOnlyMemory<byte>`
Aaronontheweb Mar 28, 2024
486f5a3
fixed `ByteStringSpec`s
Aaronontheweb Mar 28, 2024
d9c3d98
fixed framing issues
Aaronontheweb Mar 28, 2024
cd47df3
fixed compilation errors
Aaronontheweb Apr 1, 2024
31122d7
fixed issue with fuzzing spec
Aaronontheweb Apr 1, 2024
8cbb00b
added API approvals
Aaronontheweb Apr 1, 2024
53a7224
Revert "fixed framing issues"
Aaronontheweb Apr 3, 2024
1e37898
fixed frame-length encoding
Aaronontheweb Apr 3, 2024
0726e2d
added substring spec
Aaronontheweb Apr 3, 2024
abc6aa4
validated IndexOf operation
Aaronontheweb Apr 3, 2024
aca4c72
fixing up specs
Aaronontheweb Apr 3, 2024
ded7e5c
added ByteString.IndexOf method for single bytes
Aaronontheweb Apr 3, 2024
1a93896
fixing up framing
Aaronontheweb Apr 3, 2024
27ae16d
fixed some of the N+1 issues
Aaronontheweb Apr 4, 2024
79b01bd
fixed delimiter based encoding
Aaronontheweb Apr 4, 2024
f63896f
delete unused code
Aaronontheweb Apr 4, 2024
4b69e69
Merge branch 'dev' into bytestring-rewrite
Aaronontheweb Apr 11, 2024
411dcb5
Merge branch 'dev' into bytestring-rewrite
Aaronontheweb Apr 11, 2024
8a25c7a
Merge branch 'dev' into bytestring-rewrite
Aaronontheweb Apr 12, 2024
e73aef8
Merge branch 'dev' into bytestring-rewrite
Aaronontheweb May 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3766,14 +3766,17 @@ namespace Akka.IO
BigEndian = 0,
LittleEndian = 1,
}
[System.Diagnostics.DebuggerDisplayAttribute("(Count = {_count}, Buffers = {_buffers})")]
public sealed class ByteString : System.Collections.Generic.IEnumerable<byte>, System.Collections.IEnumerable, System.IEquatable<Akka.IO.ByteString>
[System.Diagnostics.DebuggerDisplayAttribute("(Count = {Count}, Buffer = {Memory})")]
public sealed class ByteString : System.IEquatable<Akka.IO.ByteString>
{
public int Count { get; }
public static Akka.IO.ByteString Empty { get; }
[System.ObsoleteAttribute("This property will be removed in future versions of Akka.NET.")]
public bool IsCompact { get; }
public bool IsEmpty { get; }
public byte this[int index] { get; }
public System.ReadOnlyMemory<byte> Memory { get; }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public Akka.IO.ByteString Compact() { }
public Akka.IO.ByteString Concat(Akka.IO.ByteString other) { }
public static Akka.IO.ByteString CopyFrom(byte[] array) { }
Expand All @@ -3785,9 +3788,13 @@ namespace Akka.IO
public static Akka.IO.ByteString CopyFrom(System.Span<byte> span, int offset, int count) { }
public static Akka.IO.ByteString CopyFrom(System.Collections.Generic.IEnumerable<System.ArraySegment<byte>> buffers) { }
public int CopyTo(byte[] buffer, int index, int count) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Memory<byte> buffer) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Memory<byte> buffer, int index, int count) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Span<byte> buffer) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Span<byte> buffer, int index, int count) { }
public override bool Equals(object obj) { }
public bool Equals(Akka.IO.ByteString other) { }
Expand All @@ -3797,7 +3804,6 @@ namespace Akka.IO
public static Akka.IO.ByteString FromBytes(System.Collections.Generic.IEnumerable<System.ArraySegment<byte>> buffers) { }
public static Akka.IO.ByteString FromString(string str) { }
public static Akka.IO.ByteString FromString(string str, System.Text.Encoding encoding) { }
public System.Collections.Generic.IEnumerator<byte> GetEnumerator() { }
public override int GetHashCode() { }
public bool HasSubstring(Akka.IO.ByteString other, int index) { }
public int IndexOf(byte b) { }
Expand All @@ -3808,12 +3814,16 @@ namespace Akka.IO
public System.ReadOnlySpan<byte> ToReadOnlySpan() { }
public override string ToString() { }
public string ToString(System.Text.Encoding encoding) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public void WriteTo(System.IO.Stream stream) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public System.Threading.Tasks.Task WriteToAsync(System.IO.Stream stream) { }
public static Akka.IO.ByteString +(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
public static bool ==(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
public static Akka.IO.ByteString op_Explicit(byte[] bytes) { }
public static byte[] op_Explicit(Akka.IO.ByteString byteString) { }
public static Akka.IO.ByteString op_Explicit(System.Memory<byte> memory) { }
public static Akka.IO.ByteString op_Explicit(System.ReadOnlyMemory<byte> memory) { }
public static bool !=(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
}
[Akka.Annotations.InternalApiAttribute()]
Expand Down
16 changes: 13 additions & 3 deletions src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.Net.verified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3756,14 +3756,17 @@ namespace Akka.IO
BigEndian = 0,
LittleEndian = 1,
}
[System.Diagnostics.DebuggerDisplayAttribute("(Count = {_count}, Buffers = {_buffers})")]
public sealed class ByteString : System.Collections.Generic.IEnumerable<byte>, System.Collections.IEnumerable, System.IEquatable<Akka.IO.ByteString>
[System.Diagnostics.DebuggerDisplayAttribute("(Count = {Count}, Buffer = {Memory})")]
public sealed class ByteString : System.IEquatable<Akka.IO.ByteString>
{
public int Count { get; }
public static Akka.IO.ByteString Empty { get; }
[System.ObsoleteAttribute("This property will be removed in future versions of Akka.NET.")]
public bool IsCompact { get; }
public bool IsEmpty { get; }
public byte this[int index] { get; }
public System.ReadOnlyMemory<byte> Memory { get; }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public Akka.IO.ByteString Compact() { }
public Akka.IO.ByteString Concat(Akka.IO.ByteString other) { }
public static Akka.IO.ByteString CopyFrom(byte[] array) { }
Expand All @@ -3775,9 +3778,13 @@ namespace Akka.IO
public static Akka.IO.ByteString CopyFrom(System.Span<byte> span, int offset, int count) { }
public static Akka.IO.ByteString CopyFrom(System.Collections.Generic.IEnumerable<System.ArraySegment<byte>> buffers) { }
public int CopyTo(byte[] buffer, int index, int count) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Memory<byte> buffer) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Memory<byte> buffer, int index, int count) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Span<byte> buffer) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public int CopyTo(ref System.Span<byte> buffer, int index, int count) { }
public override bool Equals(object obj) { }
public bool Equals(Akka.IO.ByteString other) { }
Expand All @@ -3787,7 +3794,6 @@ namespace Akka.IO
public static Akka.IO.ByteString FromBytes(System.Collections.Generic.IEnumerable<System.ArraySegment<byte>> buffers) { }
public static Akka.IO.ByteString FromString(string str) { }
public static Akka.IO.ByteString FromString(string str, System.Text.Encoding encoding) { }
public System.Collections.Generic.IEnumerator<byte> GetEnumerator() { }
public override int GetHashCode() { }
public bool HasSubstring(Akka.IO.ByteString other, int index) { }
public int IndexOf(byte b) { }
Expand All @@ -3798,12 +3804,16 @@ namespace Akka.IO
public System.ReadOnlySpan<byte> ToReadOnlySpan() { }
public override string ToString() { }
public string ToString(System.Text.Encoding encoding) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public void WriteTo(System.IO.Stream stream) { }
[System.ObsoleteAttribute("This method will be removed in future versions of Akka.NET.")]
public System.Threading.Tasks.Task WriteToAsync(System.IO.Stream stream) { }
public static Akka.IO.ByteString +(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
public static bool ==(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
public static Akka.IO.ByteString op_Explicit(byte[] bytes) { }
public static byte[] op_Explicit(Akka.IO.ByteString byteString) { }
public static Akka.IO.ByteString op_Explicit(System.Memory<byte> memory) { }
public static Akka.IO.ByteString op_Explicit(System.ReadOnlyMemory<byte> memory) { }
public static bool !=(Akka.IO.ByteString x, Akka.IO.ByteString y) { }
}
[Akka.Annotations.InternalApiAttribute()]
Expand Down
8 changes: 5 additions & 3 deletions src/core/Akka.Streams.Tests/Dsl/FlowGroupBySpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,16 +447,18 @@ public async Task GroupBy_must_work_under_fuzzing_stress_test()
await this.AssertAllStagesStoppedAsync(async () =>
{
var publisherProbe = this.CreateManualPublisherProbe<ByteString>();
var subscriber = this.CreateManualSubscriberProbe<IEnumerable<byte>>();
var subscriber = this.CreateManualSubscriberProbe<ByteString>();

var firstGroup = (Source<IEnumerable<byte>, NotUsed>)Source.FromPublisher(publisherProbe)
.GroupBy(256, element => element[0])
.Select(b => b.ToArray()) // have to convert to an array
.Select(b => b.Reverse())
.MergeSubstreams();
var secondGroup = (Source<IEnumerable<byte>, NotUsed>)firstGroup.GroupBy(256, bytes => bytes.First())
var secondGroup = (Source<ByteString, NotUsed>)firstGroup.GroupBy(256, bytes => bytes.First())
.Select(b => b.Reverse())
.Select(b => ByteString.FromBytes(b.ToArray()))
.MergeSubstreams();
var publisher = secondGroup.RunWith(Sink.AsPublisher<IEnumerable<byte>>(false), Materializer);
var publisher = secondGroup.RunWith(Sink.AsPublisher<ByteString>(false), Materializer);
publisher.Subscribe(subscriber);

var upstreamSubscription = await publisherProbe.ExpectSubscriptionAsync();
Expand Down
Loading
Loading